KIP-572 will only ship in 2.8.0. For the exception you hit, it's `max.block.ms` -- you might also look into `default.api.timeout.ms`.
In general, the relevant configs are documented in the JavaDocs of the corresponding client method. -Matthias On 11/20/20 9:11 PM, Pushkar Deole wrote: > Thanks Matthias... We are already on kafka 2.5.0, and > https://issues.apache.org/jira/browse/KAFKA-8803 mentions that these type > of issues are fixed in 2.5.0 > > Is KIP-572 planned for 2.7.0 ? > > Also, for timeout and retries, can you provide which parameters should we > configure to higher values for now? > > > On Sat, Nov 21, 2020 at 5:15 AM Matthias J. Sax <[email protected]> wrote: > >> 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"} >>> >> >
