下面是示例代码。此代码假设主题中已经有一些消息。这不会循环并等待更多消息。

Properties consumerProps = new Properties();
    consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, ip);
    consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer.class.getName());
    consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer.class.getName());
    consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "mygroup");
    KafkaConsumer consumer = new KafkaConsumer<>(consumerProps);
    ConsumerRecords<String, String> rows = consumer.poll(1000);
    Iterator<ConsumerRecord<String, String>> it = rows.iterator();
    while (it.hasNext()) {
        ConsumerRecord<String, String> row = it.next();
        MyEvent event = new MyEvent(row.value()); // transform string to event

        // process event
        runtime.sendEvent(event);
    }

我也有同样的问题。我创建了一个示例项目，你可以看看，特别是平原埃斯珀分支。
更简化的版本是：

public class KafkaExample implements Runnable {
    private String runtimeURI;

    public KafkaExample(String runtimeURI) {
        this.runtimeURI = runtimeURI;
    }

    public static void main(String[] args){
        new KafkaExample("KafkaExample").run();
    }

    @Override
    public void run() {
        Configuration configuration = new Configuration();
        configuration.getCommon().addImport(KafkaOutputDefault.class);
        configuration.getCommon().addEventType(String.class);

        Properties consumerProps = new Properties();

        // Kafka Consumer Properties
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, OffsetResetStrategy.EARLIEST.toString().toLowerCase());

        // EsperIO Kafka Input Adapter Properties
        consumerProps.put(EsperIOKafkaConfig.INPUT_SUBSCRIBER_CONFIG, Consumer.class.getName());
        consumerProps.put(EsperIOKafkaConfig.INPUT_PROCESSOR_CONFIG, InputProcessor.class.getName());
        consumerProps.put(EsperIOKafkaConfig.INPUT_TIMESTAMPEXTRACTOR_CONFIG, EsperIOKafkaInputTimestampExtractorConsumerRecord.class.getName());

        configuration.getRuntime().addPluginLoader("KafkaInput", EsperIOKafkaInputAdapterPlugin.class.getName(), consumerProps, null);

        String stmt = "@name('sampleQuery') select * from String";
        EPCompiled compiled;
        try {
            compiled = EPCompilerProvider.getCompiler().compile(stmt, new CompilerArguments(configuration));
        } catch (EPCompileException ex) {
            throw new RuntimeException(ex);
        }

        EPRuntime runtime = EPRuntimeProvider.getRuntime(runtimeURI, configuration);
        EPDeployment deployment;
        try {
            deployment = runtime.getDeploymentService().deploy(compiled, new DeploymentOptions().setDeploymentId(UUID.randomUUID().toString()));
        } catch (EPDeployException ex) {
            throw new RuntimeException(ex);
        }

        EPStatement statement = runtime.getDeploymentService().getStatement(deployment.getDeploymentId(), "sampleQuery");

        statement.addListener((newData, oldData, sta, run) -> {
            for (EventBean nd : newData) {
                System.out.println(nd.getUnderlying());
            }
        });

        while (true) {}
    }
}

public class Consumer implements EsperIOKafkaInputSubscriber {
    @Override
    public void subscribe(EsperIOKafkaInputSubscriberContext context) {
        Collection<String> collection = new ArrayList<String>();
        collection.add("input");
        context.getConsumer().subscribe(collection);
    }
}

public class InputProcessor implements EsperIOKafkaInputProcessor {
    private EPRuntime runtime;

    @Override
    public void init(EsperIOKafkaInputProcessorContext context) {
        this.runtime = context.getRuntime();
    }

    @Override
    public void process(ConsumerRecords<Object, Object> records) {
        for (ConsumerRecord record : records) {
            if (record.value() != null) {
                try {
                    runtime.getEventService().sendEventBean(record.value().toString(), "String");
                } catch (Exception e) {
                    throw e;
                }
            }
        }
    }

    public void close() {}
}