maybe you can add producer "max.block.ms" config,but you should test your broker look up some logs about leader change ,producer performance,like traffic ,produce "buffer" and "batch.size"
On Mon, Jun 6, 2022 at 6:53 PM Deepak Jain <deepak.j...@cumulus-systems.com> wrote: > Hello All, > > Please help me out in this regard as the Customer has reported this on > their production environment and waiting for our reply ASAP. > > Regards, > Deepak > > From: Deepak Jain > Sent: 02 June 2022 20:53 > To: 'users@kafka.apache.org' <users@kafka.apache.org> > Cc: 'Luke Chen' <show...@gmail.com>; Alap Patwardhan < > a...@cumulus-systems.com>; Bhushan Patil < > bhushan.pa...@cumulus-systems.com> > Subject: Random continuous TimeoutException with Topic not present on one > KafkaProducer out of many in multithreaded env > > Hello Everyone, > > We are using Kafka 2.8.1 Broker/Client system in our prod env. > > Getting following exception randomly after 1 hour or so for one Realtime > transfer from Kafka Producer to broker out of 5. (Rest 4 are working fine.) > > java.util.concurrent.ExecutionException: > org.apache.kafka.common.errors.TimeoutException: Topic realtimeImport_1 not > present in metadata after 250 ms. > at > org.apache.kafka.clients.producer.KafkaProducer$FutureFailure.<init>(KafkaProducer.java:1316) > at > org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:985) > at > org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:885) > at > org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:773) > > We are using multithreaded KafkaProducer with their each unique topic > sending data to single broker. Here, we notice that this exception comes > when we reconnect to Kafka using close() (void > org<eclipse-javadoc:%E2%98%82=hdca/D:%5C/git%5C/hdca%5C/megha%5C/app%5C/platform%5C/common%5C/tools%5C/lib%5C/kafka-clients-2.8.1.jar%3Corg>.apache<eclipse-javadoc:%E2%98%82=hdca/D:%5C/git%5C/hdca%5C/megha%5C/app%5C/platform%5C/common%5C/tools%5C/lib%5C/kafka-clients-2.8.1.jar%3Corg.apache>.kafka<eclipse-javadoc:%E2%98%82=hdca/D:%5C/git%5C/hdca%5C/megha%5C/app%5C/platform%5C/common%5C/tools%5C/lib%5C/kafka-clients-2.8.1.jar%3Corg.apache.kafka>.clients<eclipse-javadoc:%E2%98%82=hdca/D:%5C/git%5C/hdca%5C/megha%5C/app%5C/platform%5C/common%5C/tools%5C/lib%5C/kafka-clients-2.8.1.jar%3Corg.apache.kafka.clients>.producer<eclipse-javadoc:%E2%98%82=hdca/D:%5C/git%5C/hdca%5C/megha%5C/app%5C/platform%5C/common%5C/tools%5C/lib%5C/kafka-clients-2.8.1.jar%3Corg.apache.kafka.clients.producer>.KafkaProducer<eclipse-javadoc:%E2%98%82=hdca/D:%5C/git%5C/hdca%5C/megha%5C/app%5C/platform%5C/common%5C/tools%5C/lib%5C/kafka-clients-2.8.1.jar%3Corg.apache.kafka.clients.producer(KafkaProducer.class%E2%98%83KafkaProducer>.close()) > and > org<eclipse-javadoc:%E2%98%82=hdca/D:%5C/git%5C/hdca%5C/megha%5C/app%5C/platform%5C/common%5C/tools%5C/lib%5C/kafka-clients-2.8.1.jar%3Corg>.apache<eclipse-javadoc:%E2%98%82=hdca/D:%5C/git%5C/hdca%5C/megha%5C/app%5C/platform%5C/common%5C/tools%5C/lib%5C/kafka-clients-2.8.1.jar%3Corg.apache>.kafka<eclipse-javadoc:%E2%98%82=hdca/D:%5C/git%5C/hdca%5C/megha%5C/app%5C/platform%5C/common%5C/tools%5C/lib%5C/kafka-clients-2.8.1.jar%3Corg.apache.kafka>.clients<eclipse-javadoc:%E2%98%82=hdca/D:%5C/git%5C/hdca%5C/megha%5C/app%5C/platform%5C/common%5C/tools%5C/lib%5C/kafka-clients-2.8.1.jar%3Corg.apache.kafka.clients>.producer<eclipse-javadoc:%E2%98%82=hdca/D:%5C/git%5C/hdca%5C/megha%5C/app%5C/platform%5C/common%5C/tools%5C/lib%5C/kafka-clients-2.8.1.jar%3Corg.apache.kafka.clients.producer>.KafkaProducer<eclipse-javadoc:%E2%98%82=hdca/D:%5C/git%5C/hdca%5C/megha%5C/app%5C/platform%5C/common%5C/tools%5C/lib%5C/kafka-clients-2.8.1.jar%3Corg.apache.kafka.clients.producer(KafkaProducer.class%E2%98%83KafkaProducer>.KafkaProducer<Object<eclipse-javadoc:%E2%98%82=hdca/D:%5C/git%5C/hdca%5C/megha%5C/app%5C/platform%5C/common%5C/tools%5C/lib%5C/kafka-clients-2.8.1.jar%3Corg.apache.kafka.clients.producer(KafkaProducer.class%E2%98%83KafkaProducer~KafkaProducer~Ljava.util.Properties;%E2%98%82java.lang.Object>, > Object<eclipse-javadoc:%E2%98%82=hdca/D:%5C/git%5C/hdca%5C/megha%5C/app%5C/platform%5C/common%5C/tools%5C/lib%5C/kafka-clients-2.8.1.jar%3Corg.apache.kafka.clients.producer(KafkaProducer.class%E2%98%83KafkaProducer~KafkaProducer~Ljava.util.Properties;%E2%98%82java.lang.Object>>(Properties<eclipse-javadoc:%E2%98%82=hdca/D:%5C/git%5C/hdca%5C/megha%5C/app%5C/platform%5C/common%5C/tools%5C/lib%5C/kafka-clients-2.8.1.jar%3Corg.apache.kafka.clients.producer(KafkaProducer.class%E2%98%83KafkaProducer~KafkaProducer~Ljava.util.Properties;%E2%98%82java.util.Properties> > properties) calls. Not sure whether this is the culprit or not. > > Due to this exception the realtime resources are not getting transfer to > Kafka Consumer. We are using following config on Kafka Broker: > > group.initial.rebalance.delay.ms=0 > listeners=SASL_PLAINTEXT://0.0.0.0:9092 > log.retention.minutes=15 > delete.topic.enable=true > auto.create.topics.enable=true > security.inter.broker.protocol=SASL_PLAINTEXT > sasl.mechanism.inter.broker.protocol=PLAIN > sasl.enabled.mechanisms=PLAIN > security.protocol=SASL_PLAINTEXT > authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer > allow.everyone.if.no.acl.found=true > zookeeper.set.acl=true > > Can anyone please help us in finding the root cause for it? > > Regards, > Deepak Jain > Cumulus Systems >
