Dead loop for connector state

Hello team,

I am using Camel-kafka-s3 connector version 3.3.0 and apache camel 3.7.0. I have been using for more than a year and had never seen this issue. We have kafka cluster strimi 0.25.0 for kafka 2.8.0.

I noticed that for some issue (most probably custom broker certificate expired) on kafka-cluster, our kafka connector went on dead loop of re-trying to connect to broker. Our cluster came back in 10-15 mins but the kafka-connector process never came back to a healthy status. Here is what I see in logs:

2022-02-28 02:11:33,287 WARN [Worker clientId=connect-1, groupId=s3-connector] Didn't reach end of config log quickly enough (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
java.util.concurrent.TimeoutException: Timed out waiting for future
at org.apache.kafka.connect.util.ConvertingFutureCallback.get(ConvertingFutureCallback.java:106)
at org.apache.kafka.connect.storage.KafkaConfigBackingStore.refresh(KafkaConfigBackingStore.java:441)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.readConfigToEnd(DistributedHerder.java:1188)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.tick(DistributedHerder.java:342)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.run(DistributedHerder.java:316)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:829)
2022-02-28 02:16:45,296 INFO [Worker clientId=connect-1, groupId=s3-connector] Reading to end of config log; current config state offset: 2233 (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]

Follwing the log from where exception started:

2022-02-27 19:18:30,203 INFO [AdminClient clientId=adminclient-8] Failed authentication with kafka-cluster-kafka-15.kafka-cluster-kafka-brokers.mdic-strimzi-pro.svc/100.80.52.142 (Token validation failed: Unknown signing key (kid:{keyid}) (ErrId: 26617d8d)) (org.apache.kafka.common.network.Selector) [kafka-admin-client-thread | adminclient-8]
2022-02-27 19:18:30,203 ERROR [AdminClient clientId=adminclient-8] Connection to node 15 (kafka-cluster-kafka-15.kafka-cluster-kafka-brokers.mdic-strimzi-pro.svc/100.80.52.142:9093) failed authentication due to: Token validation failed: Unknown signing key (kid:{keyid}) (ErrId: 26617d8d) (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-8]
2022-02-27 19:18:30,203 WARN [AdminClient clientId=adminclient-8] Metadata update failed due to authentication error (org.apache.kafka.clients.admin.internals.AdminMetadataManager) [kafka-admin-client-thread | adminclient-8]
org.apache.kafka.common.errors.SaslAuthenticationException: Token validation failed: Unknown signing key (kid:m45ykTYExWg7_fe2lwAqTCHevnt-1c064bM9vRSw5V4) (ErrId: 26617d8d)
2022-02-27 19:18:30,218 INFO [AdminClient clientId=adminclient-8] Metadata update failed (org.apache.kafka.clients.admin.internals.AdminMetadataManager) [kafka-admin-client-thread | adminclient-8]
org.apache.kafka.common.errors.TimeoutException: Call(callName=fetchMetadata, deadlineMs=1645989510217, tries=1, nextAllowedTryMs=1645989510318) timed out at 1645989510218 after 1 attempt(s)
Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting to send the call. Call: fetchMetadata
2022-02-27 19:18:30,318 INFO [AdminClient clientId=adminclient-8] Metadata update failed (org.apache.kafka.clients.admin.internals.AdminMetadataManager) [kafka-admin-client-thread | adminclient-8]
org.apache.kafka.common.errors.TimeoutException: Call(callName=fetchMetadata, deadlineMs=1645989510317, tries=1, nextAllowedTryMs=1645989510418) timed out at 1645989510318 after 1 attempt(s)
Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting to send the call. Call: fetchMetadata
2022-02-27 19:18:30,418 INFO [AdminClient clientId=adminclient-8] Metadata update failed (org.apache.kafka.clients.admin.internals.AdminMetadataManager) [kafka-admin-client-thread | adminclient-8]
org.apache.kafka.common.errors.TimeoutException: Call(callName=fetchMetadata, deadlineMs=1645989510417, tries=1, nextAllowedTryMs=1645989510518) timed out at 1645989510418 after 1 attempt(s)
Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting to send the call. Call: fetchMetadata
2022-02-27 19:18:30,518 INFO [AdminClient clientId=adminclient-8] Metadata update failed (org.apache.kafka.clients.admin.internals.AdminMetadataManager) [kafka-admin-client-thread | adminclient-8]
org.apache.kafka.common.errors.TimeoutException: Call(callName=fetchMetadata, deadlineMs=1645989510517, tries=1, nextAllowedTryMs=1645989510618) timed out at 1645989510518 after 1 attempt(s)
Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting to send the call. Call: fetchMetadata

I have few following questions:

1. Why didn’t the connector die and restart the process?
2. Is there a way we can propagate the Admin client exception in the application?
3. If not, How can I monitor my application health?

Thank you.