Matthias,
We recently ran into an issue where kafka brokers upgraded (i guess it was
rolling update) for Aiven business plan 4 to plan 8. This involves change
to cpu, memory and storage for each broker.
Since this should be rolling upgrade, we expected services to survive,
however in one service we saw streams ran into Error with below exception:
Few questions:
1. If a broker goes down, the kafka streams client should handle internally
and connect to available broker since we have topic with replicas equal to
no. of brokers. Is this correct?
2. the below error says timeout expired while awaiting InitProducerId..
what does this signify and why would this timeout occur when there will be
other brokers up and running?
{"@timestamp":"2020-11-16T13:42:31.110+00:00","@version":"1","message":"Unexpected
exception in stream
processing.","logger_name":"com.avaya.analytics.filter.ApplicationConfig","thread_name":"analytics-event-filter-StreamThread-1","level":"ERROR","level_value":40000,"stack_trace":"org.apache.kafka.streams.errors.StreamsException:
stream-thread [analytics-event-filter-StreamThread-1] Failed to
rebalance.\n\tat
org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:862)\n\tat
org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:753)\n\tat
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:697)\n\tat
org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:670)\nCaused
by: org.apache.kafka.streams.errors.StreamsException: stream-thread
[analytics-event-filter-StreamThread-1] task [1_0] Failed to initialize
task 1_0 due to timeout.\n\tat
org.apache.kafka.streams.processor.internals.StreamTask.initializeTransactions(StreamTask.java:923)\n\tat
org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:206)\n\tat
org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:115)\n\tat
org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:352)\n\tat
org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:310)\n\tat
org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.createTasks(StreamThread.java:295)\n\tat
org.apache.kafka.streams.processor.internals.TaskManager.addNewActiveTasks(TaskManager.java:160)\n\tat
org.apache.kafka.streams.processor.internals.TaskManager.createTasks(TaskManager.java:120)\n\tat
org.apache.kafka.streams.processor.internals.StreamsRebalanceListener.onPartitionsAssigned(StreamsRebalanceListener.java:77)\n\tat
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsAssigned(ConsumerCoordinator.java:278)\n\tat
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:419)\n\tat
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:439)\n\tat
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:358)\n\tat
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:497)\n\tat
org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1274)\n\tat
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1238)\n\tat
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1173)\n\tat
brave.kafka.clients.TracingConsumer.poll(TracingConsumer.java:89)\n\tat
brave.kafka.clients.TracingConsumer.poll(TracingConsumer.java:83)\n\tat
org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:853)\n\t...
3 common frames omitted\nCaused by:
org.apache.kafka.common.errors.TimeoutException: Timeout expired after
60000milliseconds while awaiting InitProducerId\n"}
