0

I am trying to join two Kafka Stream DSL KTable using:

KTable<String, String> source = builder.table("stream-source");
KTable<String, String> target = builder.table("stream-target");
source.join(target, new ValueJoiner<String, String, String>() {
    public String apply(String value1, String value2) {
        return value1 + ":" + value2;
    }
});

I have made sure that both the keys and the values are not null:

Producer<String, String> producer = new KafkaProducer<String, String>(props);
for(int i = 0; i < PERSONS_SOURCE.length; i++) {
    producer.send(new ProducerRecord<String, String>("stream-source",     Long.toString(i + 1L), PERSONS_SOURCE[i]));
}
for(int i = 0; i < PERSONS_TARGET.length; i++) {
    producer.send(new ProducerRecord<String, String>("stream-target", Long.toString(i + 1L), PERSONS_TARGET[i]));
}
producer.close();

But the application reports that there is a null pointer in the RocksDB layer concerning the partition.

[2016-07-17 21:58:04,682] ERROR User provided listener org.apache.kafka.streams.processor.internals.StreamThread$1 for group streams-persons2 failed on partition assignment (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator) java.lang.NullPointerException at org.rocksdb.RocksDB.put(RocksDB.java:432) at org.apache.kafka.streams.state.internals.RocksDBStore.putInternal(RocksDBStore.java:299) at org.apache.kafka.streams.state.internals.RocksDBStore.access$200(RocksDBStore.java:62) at org.apache.kafka.streams.state.internals.RocksDBStore$3.restore(RocksDBStore.java:206) at org.apache.kafka.streams.processor.internals.ProcessorStateManager.restoreActiveState(ProcessorStateManager.java:245) at org.apache.kafka.streams.processor.internals.ProcessorStateManager.register(ProcessorStateManager.java:210) at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.register(ProcessorContextImpl.java:116) at org.apache.kafka.streams.state.internals.RocksDBStore.init(RocksDBStore.java:202)

4

1 回答 1

1

发现问题是由于在应用程序代码中创建的流而不是使用命令：-

kafka-topics --create --topic stream-a --replication-factor 1 --partitions 1

似乎连接需要分区信息才能工作。

于 2016-07-18T19:08:44.600 回答