I am trying to use Apache Kafka in a scenario with 24 nodes where the connection between the nodes is characterized by packet loss and long round trip times (~1 to 4 seconds).
I am using KAFKA as a pub/sub broker, so all the consumers and producers receive information from a centralized broker.
I made sure to assign a unique group ID to each consumer/publisher.
If I run my code in an Enterprise network (very small RTT, no packet loss) everything works well but if I try to run my code in the constrained network, I can observe that the messages get delivered to the central broker but are then not bounced to other consumers. Actually each node receives most of the first 2 messages but then it stops receiving new ones.
I am not detecting any error or exception so far.
I also deployed a consumer in the broker machine and it seems to be receiving the messages fine.
I also tried to use the console consumer utility from one of the nodes and it too was not able to receive messages so I started wondering if the problem could be related to the broker configuration.
I have other TCP based protocols (NATS, MQTT, REDIS) and they work just fine.
I wonder if there is any specific configuration I have to add to make KAFKA more fault tolerant. These are my Consumer and Producer configurations:
Consumer:
properties.put (ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, _server);
properties.put (ConsumerConfig.GROUP_ID_CONFIG, _groupID);
properties.put (ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
KafkaBroker.KEY_DESERIALIZER);
properties.put (ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
KafkaBroker.VALUE_DESERIALIZER);
properties.put ("enable.auto.commit", "true");
properties.put ("auto.commit.interval.ms", "1000");
properties.put ("session.timeout.ms", "30000");
Producer:
properties.put ("acks", "all");
properties.put (ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, _server);
properties.put (ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
KafkaBroker.KEY_SERIALIZER);
properties.put (ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
KafkaBroker.VALUE_SERIALIZER);
//If the request fails, the producer can automatically retry,
properties.put (ProducerConfig.RETRIES_CONFIG, 0);
properties.put (ProducerConfig.CLIENT_ID_CONFIG, _clientID);