Hi team, I have a cluster which has a realtime table ingesting from kafka. everytime I delete the segments, the connection between the table and the kafka breaks Can any one help me understand the root cause here.

2025/04/10 09:34:36.948 INFO [HelixInstanceDataManager] [HelixTaskExecutor-message_handle_thread_21] Deleted segment: ruleset_evaluation_shadow_test_realtime_2__2__47__202
50410T0520Z from table: ruleset_evaluation_shadow_test_realtime_2_REALTIME
2025/04/10 09:34:37.177 INFO [RealtimeSegmentDataManager_ruleset_evaluation_shadow_test_realtime_2__2__49__20250410T0716Z] [ruleset_evaluation_shadow_test_realtime_2__2__4
9__20250410T0716Z] Recreating stream consumer for topic partition ruleset_evaluation_shadow_test_realtime_2_REALTIME-topic.ruleset.evaluation.shadow.test-2, reason: Too ma
ny transient errors
2025/04/10 09:34:37.177 ERROR [KafkaConsumer] [ruleset_evaluation_shadow_test_realtime_2__2__49__20250410T0716Z] [Consumer clientId=ruleset_evaluation_shadow_test_realtime
_2_REALTIME-topic.ruleset.evaluation.shadow.test-2, groupId=msk_consumer] Failed to close coordinator
org.apache.kafka.common.errors.InterruptException: java.lang.InterruptedException
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeThrowInterruptException(ConsumerNetworkClient.java:520) ~[pinot-all-1.4.0-SNAPSHOT-jar-with-dependencies.jar:1.4.0-SNAPSHOT-eb9c759344502969c80e3e9ec00fe67bd24d2965]
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:281) ~[pinot-all-1.4.0-SNAPSHOT-jar-with-dependencies.jar:1.4.0-SNAPSHOT-eb9c759344502969c80e3e9ec00fe67bd24d2965]
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:236) ~[pinot-all-1.4.0-SNAPSHOT-jar-with-dependencies.jar:1.4.0-SNAPSHOT-eb9c759344502969c80e3e9ec00fe67bd24d2965]
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:215) ~[pinot-all-1.4.0-SNAPSHOT-jar-with-dependencies.jar:1.4.0-SNAPSHOT-eb9c759344502969c80e3e9ec00fe67bd24d2965]
        at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:245) ~[pinot-all-1.4.0-SNAPSHOT-jar-with-dependencies.jar:1.4.0-SNAPSHOT-eb9c759344502969c80e3e9ec00fe67bd24d2965]
        at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:1026) ~[pinot-all-1.4.0-SNAPSHOT-jar-with-dependencies.jar:1.4.0-SNAPSHOT-eb9c759344502969c80e3e9ec00fe67bd24d2965]
        at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.maybeAutoCommitOffsetsSync(ConsumerCoordinator.java:1087) ~[pinot-all-1.4.0-SNAPSHOT-jar-with-dependencies.jar:1.4.0-SNAPSHOT-eb9c759344502969c80e3e9ec00fe67bd24d2965]
        at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.close(ConsumerCoordinator.java:919) ~[pinot-all-1.4.0-SNAPSHOT-jar-with-dependencies.jar:1.4.0-SNAPSHOT-eb9c759344502969c80e3e9ec00fe67bd24d2965]
        at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2366) ~[pinot-all-1.4.0-SNAPSHOT-jar-with-dependencies.jar:1.4.0-SNAPSHOT-eb9c759344502969c80e3e9ec00fe67bd24d2965]
        at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2333) ~[pinot-all-1.4.0-SNAPSHOT-jar-with-dependencies.jar:1.4.0-SNAPSHOT-eb9c759344502969c80e3e9ec00fe67bd24d2965]
        at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2283) ~[pinot-all-1.4.0-SNAPSHOT-jar-with-dependencies.jar:1.4.0-SNAPSHOT-eb9c759344502969c80e3e9ec00fe67bd24d2965]
        at org.apache.pinot.plugin.stream.kafka20.KafkaPartitionLevelConnectionHandler.close(KafkaPartitionLevelConnectionHandler.java:118) ~[pinot-all-1.4.0-SNAPSHOT-jar-with-dependencies.jar:1.4.0-SNAPSHOT-eb9c759344502969c80e3e9ec00fe67bd24d2965]
        at org.apache.pinot.core.data.manager.realtime.RealtimeSegmentDataManager.closePartitionGroupConsumer(RealtimeSegmentDataManager.java:1256) ~[pinot-all-1.4.0-SNAPSHOT-jar-with-dependencies.jar:1.4.0-SNAPSHOT-eb9c759344502969c80e3e9ec00fe67bd24d2965]
        at org.apache.pinot.core.data.manager.realtime.RealtimeSegmentDataManager.recreateStreamConsumer(RealtimeSegmentDataManager.java:1860) ~[pinot-all-1.4.0-SNAPSHOT-jar-with-dependencies.jar:1.4.0-SNAPSHOT-eb9c759344502969c80e3e9ec00fe67bd24d2965]
        at org.apache.pinot.core.data.manager.realtime.RealtimeSegmentDataManager.handleTransientStreamErrors(RealtimeSegmentDataManager.java:436) ~[pinot-all-1.4.0-SNAPSHOT-jar-with-dependencies.jar:1.4.0-SNAPSHOT-eb9c759344502969c80e3e9ec00fe67bd24d2965]
        at org.apache.pinot.core.data.manager.realtime.RealtimeSegmentDataManager.consumeLoop(RealtimeSegmentDataManager.java:484) ~[pinot-all-1.4.0-SNAPSHOT-jar-with-dependencies.jar:1.4.0-SNAPSHOT-eb9c759344502969c80e3e9ec00fe67bd24d2965]
        at org.apache.pinot.core.data.manager.realtime.RealtimeSegmentDataManager$PartitionConsumer.run(RealtimeSegmentDataManager.java:766) ~[pinot-all-1.4.0-SNAPSHOT-jar-with-dependencies.jar:1.4.0-SNAPSHOT-eb9c759344502969c80e3e9ec00fe67bd24d2965]
        at java.base/java.lang.Thread.run(Thread.java:1583) [?:?]
Caused by: java.lang.InterruptedException
        ... 18 more
2025/04/10 09:34:37.177 INFO [Metrics] [ruleset_evaluation_shadow_test_realtime_2__2__49__20250410T0716Z] Metrics scheduler closed
2025/04/10 09:34:37.177 INFO [Metrics] [ruleset_evaluation_shadow_test_realtime_2__2__49__20250410T0716Z] Closing reporter org.apache.kafka.common.metrics.JmxReporter
2025/04/10 09:34:37.177 INFO [Metrics] [ruleset_evaluation_shadow_test_realtime_2__2__49__20250410T0716Z] Metrics reporters closed
2025/04/10 09:34:37.178 INFO [AppInfoParser] [ruleset_evaluation_shadow_test_realtime_2__2__49__20250410T0716Z] App info kafka.consumer for ruleset_evaluation_shadow_test_realtime_2_REALTIME-topic.ruleset.evaluation.shadow.test-2 unregistered
what is happening in these logs

If you are deleting the consuming segment then this is expected.

Okay, thanks for the information.

any segment can be deleted

Deleting manually is not a production usecase, I did it since I introduced bloom filter in the updated config and I though of deleting the older segments since the data was not required. Moving forward, I'll introduce some TTL

ok