Apache Ignite Kafka连接问题

Question

我正在尝试对Kafka消息流进行流处理和CEP。为此，我选择Apache Ignite首先实现一个原型。但是,我无法连接到队列：

使用KAFKA2.11-0.10.1.0 apache-ignite-fabric-1.8.0-bin

Kafka工作正常，我用一个消费者测试了它。然后启动ignite，然后在spring boot命令行应用程序中运行following。

    KafkaStreamer<String, String, String> kafkaStreamer = new KafkaStreamer<>();

    Ignition.setClientMode(true);

    Ignite ignite = Ignition.start();

    Properties settings = new Properties();
    // Set a few key parameters
    settings.put("bootstrap.servers", "localhost:9092");
    settings.put("group.id", "test");
    settings.put("zookeeper.connect", "localhost:2181");
    settings.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    settings.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    settings.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    settings.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

    // Create an instance of StreamsConfig from the Properties instance
    kafka.consumer.ConsumerConfig config = new ConsumerConfig(settings);

    IgniteCache<String, String> cache = ignite.getOrCreateCache("myCache");

    try (IgniteDataStreamer<String, String> stmr = ignite.dataStreamer("myCache")) {
        // allow overwriting cache data
        stmr.allowOverwrite(true);

        kafkaStreamer.setIgnite(ignite);
        kafkaStreamer.setStreamer(stmr);

        // set the topic
        kafkaStreamer.setTopic("test");

        // set the number of threads to process Kafka streams
        kafkaStreamer.setThreads(1);

        // set Kafka consumer configurations
        kafkaStreamer.setConsumerConfig(config);

        // set decoders
        StringDecoder keyDecoder = new StringDecoder(null);
        StringDecoder valueDecoder = new StringDecoder(null);

        kafkaStreamer.setKeyDecoder(keyDecoder);
        kafkaStreamer.setValueDecoder(valueDecoder);

        kafkaStreamer.start();
    } finally {
        kafkaStreamer.stop();
    }

当应用程序启动时，我得到

2017-02-23 10:25:23.409警告1388---[main]Kafka.utils.VerifiableProperties：属性bootstrap.servers无效2017-02-23 10:25:23.410信息1388---[main]Kafka.utils.VerifiableProperties：属性组ID被重写以测试2017-02-23 10:25:23.410警告1388---[main]Kafka.utils.VerifiableProperties：属性密钥.反序列化程序无效2017-02-23 10:25:23.411警告1388---[main]Kafka.utils.VerifiableProperties：属性密钥.序列化程序无效2017-02-23 10:25:23.411警告1388---[main]Kafka.utils.Verifi ableProperties：属性值.反序列化程序无效2017-02-23 10:25:23.411警告1388---[main]Kafka.utils.VerifiableProperties：属性值.序列化程序无效2017-02-23 10:25:23.411 INFO 1388---[main]Kafka.utils.VerifiableProperties：属性ZookeePer.Connect被覆盖到LocalHost：2181

然后

2017-02-23 10:25:24.057警告1388---[r-finder-thread]kafka.client.clientutils$:从代理[BrokerEndpoint(0,user.local,9092)]为主题[set(test)]提取相关id为0的主题元数据失败

java.nhtml" target="_blank">io.channels.ClosedChannelException：null在kafka.network.blockingchannel.send（blockingchannel.scala:110)~[kafka2.11-0.10.0.1.jar:na]在kafka.producer.syncproducer.liftedtree1$1(Syncproducer.scala:80)~[kafka2.11-0.10.0.1.jar:na]在kafka.producer.syncproducer.scala:80)~[kafka2.11-0.10.0.1.jar:na]在kafka.producer.syncproducer.kafka$producer.syncproducer.dosend（124)~[kafka2.11-0.10.0.1.jar:na]在kafka.client.clientutils.scala:59)[kafka2.11-0.10.0.1.jar:na]在kafka.client.clientutils.scala:94)[kafka2.11-0.10.0.1.jar:na]在kafka.client.client.clientutils.fetchtopicmetadata(Clientutils.scala:94)[kafka2.11-0.10.0.1.jar:na]在kafka.consumer.consumerfetchermanager$leaderfinderthread.doWork(consumerfetchermanager.scala:66)[kafka2.11-0.10.0.1.jar:na]在.run(SutdownableThread.Scala:63)[kafka2.11-0.10.0.1.jar:na]

并且从队列中读取也不起作用。有没有人知道怎么解决这个问题？

编辑：如果我注释finally块的内容，则会出现以下错误

[2M2017-02-27 16:42:27.780[0；39M[31Merror[0；39M[35M29946[0；39M[2M---[0；39M[2M[2M[3M]池-3-线程-1][0；39M[36M[0；39M[2M:[0；39M]39M消息被忽略，原因是错误[msg=messageandmetadata（test,0,Message（magic=1,attributes=0,CreateTime=-1,crc=2558126716,key=java.nio.heapbytebuffer[pos=0 lim=1 cap=79],payload=java.nio.heapbytebuffer[pos=0 lim=74 CAP=74]),15941704,Kafka.Serializer.StringDecoder@74A96647,Kafka.Serializer.StringDecoder@42849D34,-1,CreateTime)]

IllegalStateException：数据流已经关闭。在org.apache.ignite.internal.processors.DataStreamer.DataStreamerImpl.Enterbusy（DataStreamerImpl.java:401)~[ignite-core-1.8.0.jar:1.8.0]在org.apache.ignite.internal.processors.DataStreamer.DataStreamerImpl.AddDataInternal（DataStreamerImpl.java:613)~[ignite-core-1.8.0.jar:1.8.0]在org.apache.ignite.internal.processors.DataStreamer.DataStreamer.AddData（DataStreamerImpl.java:667)~[ignite-core-1.8.0.jar:1.8.0]在.ignite.stream.kafka.kafkastreamer$1.run(kafkastreamer.java:180)~[ignite-kafka-1.8.0.jar:1.8.0]在java.util.concurrent.executors$runnableadapter.call(executors.java:511)[NA:1.8.0_111]在java.util.concurrent.futuretask.run(futuretask.java:266)[NA:1.8.0_111]在java.util.concurrent.threadpoolexecutor.runworker(threadpoolexecutor.1142)[NA:1.8.0_111]在java.util.concurrent.threadpoolexecutor$works er.run(ThreadPoolExecutor.java:617)[NA:1.8.0_111]位于java.lang.Thread.run(Thread.java:745)[NA:1.8.0_111]

谢谢！

Answer 1

我认为发生这种情况是因为KafkastReamer在启动后立即关闭(KafkastReamer.stop()调用finally块）。kafkastreamer.start()不是同步的，它只是从Kafka中旋转出要消耗的线程并退出。