Yes, if brokers are upgraded via rolling bounce, and the embedded clients are configured with large enough timeouts and retries, they should just fail over to running brokers if a single broker is bounced.
If you get a timeout exception, than KafkaStreams dies atm -- we have KIP-572 in-flight that will improve the situation by adding one more retry layer within KafkaStreams itself. For now, you would need to increase the corresponding client timeouts to avoid that the client throws a timeout exception. There is however https://issues.apache.org/jira/browse/KAFKA-8803 that you could have hit, too. -Matthias On 11/18/20 7:05 AM, Pushkar Deole wrote: > 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"} >
