Kafka Streams: ошибка раздела State Store - PullRequest
Новая
       9

Kafka Streams: ошибка раздела State Store

0 голосов
/ 15 мая 2018

Определен пользовательский магазин, для использования в пользовательском Transformer (ссылка ниже).

https://github.com/apache/kafka/blob/trunk/streams/examples/src/main/java/org/apache/kafka/streams/examples/wordcount/WordCountProcessorDemo.java 

public class KafkaStream {
    public static void main(String[] args) {

        StateStoreSupplier houseStore = Stores.create("HOUSE").withKeys(Serdes.String()).withValues(houseSerde).persistent().build();
        KStreamBuilder kstreamBuilder = new KStreamBuilder();
        kstreamBuilder.addStateStore(houseStore);
        .
        .
        .

        KStream<String, String> testStream = kstreamBuilder.stream(Serdes.String(), Serdes.String(), "test");
        testStream.transform(HourlyDetail::new, houseStore.name());
        .
        .
        .
    }
}

class HouseDetail implements Transformer<String, String, KeyValue<String, House>> {
    @SuppressWarnings("unchecked")
    @Override
    public void init(ProcessorContext context) {
        this.usageStore = (KeyValueStore<String, House>) context.getStateStore("HOUSE");
    }
    .
    .
    .
}


Я получаю следующее исключение. Не уверен, почему внутренняя тема «test_01-HOUSE-changelog» создается с одним разделом и одной репликацией, в отличие от 2 разделов в исходном разделе «test». Чего здесь не хватает? 

[2018-05-14 23:38:09,391] ERROR stream-thread [StreamThread-1] Failed to create an active task 0_1:  (org.apache.kafka.streams.processor.internals.StreamThread:666)
org.apache.kafka.streams.errors.StreamsException: task [0_1] Store HOUSE's change log (test_01-HOUSE-changelog) does not contain partition 1
        at org.apache.kafka.streams.processor.internals.ProcessorStateManager.register(ProcessorStateManager.java:185)
        at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.register(ProcessorContextImpl.java:123)
        at org.apache.kafka.streams.state.internals.RocksDBStore.init(RocksDBStore.java:169)
        at org.apache.kafka.streams.state.internals.MeteredKeyValueStore.init(MeteredKeyValueStore.java:85)
        at org.apache.kafka.streams.processor.internals.AbstractTask.initializeStateStores(AbstractTask.java:81)
        at org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:119)
        at org.apache.kafka.streams.processor.internals.StreamThread.createStreamTask(StreamThread.java:633)
        at org.apache.kafka.streams.processor.internals.StreamThread.addStreamTasks(StreamThread.java:660)
        at org.apache.kafka.streams.processor.internals.StreamThread.access$100(StreamThread.java:69)
        at org.apache.kafka.streams.processor.internals.StreamThread$1.onPartitionsAssigned(StreamThread.java:124)
        at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:228)
        at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:313)
        at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:277)
        at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:259)
        at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1013)
        at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:979)
        at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:407)
        at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:242)



$ ./kafka-topics.sh --zookeeper localhost:2181 --topic test --describe                                                      
Topic:test  PartitionCount:2        ReplicationFactor:3     Configs:
   Topic: test Partition: 0    Leader: 1001    Replicas: 1001,1002,1003        Isr: 1002,1001,1003
   Topic: test Partition: 1    Leader: 1002    Replicas: 1002,1003,1001        Isr: 1002,1001,1003

$ ./kafka-topics.sh --zookeeper localhost:2181 --topic test_01-HOUSE-changelog --describe                                    
Topic:test_01-HOUSE-changelog        PartitionCount:1        ReplicationFactor:1     Configs:
   Topic: test_01-HOUSE-changelog       Partition: 0    Leader: 1001    Replicas: 1001  Isr: 1001

Исключение после отключения автоматического создания темы 

[2018-05-17 14:25:41,114] ERROR stream-thread [StreamThread-1] Failed to create an active task 0_0:  (org.apache.kafka.streams.processor.internals.StreamThread:666)
org.apache.kafka.streams.errors.StreamsException: task [0_0] Could not find partition info for topic: test_01-HOUSE-changelog
        at org.apache.kafka.streams.processor.internals.ProcessorStateManager.register(ProcessorStateManager.java:174)
        at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.register(ProcessorContextImpl.java:123)
        at org.apache.kafka.streams.state.internals.RocksDBStore.init(RocksDBStore.java:169)
        at org.apache.kafka.streams.state.internals.MeteredKeyValueStore.init(MeteredKeyValueStore.java:85)
        at org.apache.kafka.streams.processor.internals.AbstractTask.initializeStateStores(AbstractTask.java:81)
        at org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:119)
        at org.apache.kafka.streams.processor.internals.StreamThread.createStreamTask(StreamThread.java:633)
        at org.apache.kafka.streams.processor.internals.StreamThread.addStreamTasks(StreamThread.java:660)
        at org.apache.kafka.streams.processor.internals.StreamThread.access$100(StreamThread.java:69)
        at org.apache.kafka.streams.processor.internals.StreamThread$1.onPartitionsAssigned(StreamThread.java:124)
        at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:228)
        at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:313)
        at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:277)
        at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:259)
        at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1013)
        at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:979)
        at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:407)
        at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:242)

1 Ответ

0 голосов
/ 15 мая 2018

Если тема существует с одним разделом, Kafka Streams не будет автоматически изменять количество разделов. Непонятно, почему тема была создана с одним разделом из информации, которую вы предоставляете. Возможно, у вашей входной темы были одни разделы, когда вы запускали свое приложение в первый раз, а позже добавляли вторые разделы к входной теме.

Вам необходимо очистить приложение, используя инструмент сброса приложения, как описано в документации (обратите внимание, что это двухэтапный процесс): https://docs.confluent.io/current/streams/developer-guide/app-reset-tool.html

...