Kafka notifications: `connections.max.idle.ms` and retry policy

[fgalan]

As a continuation of issue #4666, the producer created in KafkaConnectionManager::getConnection() is not currently setting the connections.max.idle.ms.

See more details at #4666 (comment)

Kafka producers have the configuration parameter connections.max.idle.ms to close inactive sockets (without destroying the producer). This releases network resources without destroying the producer. If the producer needs to send a message after the socket is closed, it automatically opens a new connection. Kafka does not “reconnect” the same closed TCP connection; it creates a new one (to the same broker). However, the producer object (rd_kafka_t) remains the same, keeping its configuration and internal state. It does not free the producer’s memory.

What should we do in Orion? Avoid destroying producers, or be very conservative about doing so—destroy them only when strictly necessary.

When would we need to destroy the producer in Orion?

If we need to apply credential changes, which is not implemented at the moment.

Inactive clusters (long periods) — cleanup should be conservative (30+ days).

This leads to the following issue:

1. Apply the connections.max.idle.ms configuration to the Kafka producer (improvement)

As a side setting, the retry policy should be also set.

[orianar]

Check the following parameters for producer retries in librdkafka:

retries → alias of message.send.max.retries: how many times to retry sending in the event of retryable broker errors.
docs.huihoo.com. https://docs.huihoo.com/apache/kafka/confluent/4.0/clients/librdkafka/CONFIGURATION_8md.html?utm_source=chatgpt.com

retry.backoff.ms (and retry.backoff.max.ms) → wait between retries; controls the backoff (linear/exponential depending on the relationship between the two).
docs.confluent.io https://docs.confluent.io/platform/current/clients/librdkafka/html/md_INTRODUCTION.html?utm_source=chatgpt.com

delivery.timeout.ms → total limit for completing delivery; if reached, it fails even if retries are still available.
docs.confluent.io. https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html?utm_source=chatgpt.com

request.timeout.ms → maximum time per request to the broker (each individual attempt). https://stackoverflow.com/questions/64390761/how-kafka-retries-works-with-request-timeout?utm_source=chatgpt.com

Practical summary: adjust retries/message.send.max.retries and retry.backoff.ms; and remember that delivery.timeout.ms sets the global limit, so if it is too low, it will “cut off” before exhausting the retries.