Unexpected behavior by fluentd-collector from one Elastic search index to kafka

[phanitejak]

Describe the bug

We have implemented fluentd-kafka plugin to collect information from kafka topics and post the formatted data to Elastic search database. However, recently, after upgrading from kafka-6.0.0-1090 to kafka-7.0.0-1250 (these RPMs are part of our organization's internal packaging), we have observed that the custom micro-service alarm-collector build using fluentd-kafka plugin 0.16.0 version, was able to process many indexes but one of them - alarmhistoryindex - keeps failing to connect to kafka. But after restarting the micro-service pods, the connection was successful. We request a possible explanation for this. This feels some bug to us - if true, may we know the possible fix please? Thank you all in advance. alarm-collector-belk-fluentd-statefulset-0(1).log

To Reproduce

issue seems sporadic in nature. Upgrading kafka seems to have impact on fluentd gem

Expected behavior

all elastic search indexes must be attached once kafka is up and running

Your Environment

- Fluentd version: 1.12.1
- TD Agent version:
- fluent-plugin-kafka version: 0.16.0
- ruby-kafka version:
- Operating system: RHEL 7.9
- Kernel version:

Your Configuration

fluentd-conf.tar.gz

Your Error Log

log file attached

Additional context

No response