wasniksudesh commented on PR #17350: URL: https://github.com/apache/iceberg/pull/17350#issuecomment-6234925376
Hey! I think this PR will greatly benefit from https://github.com/apache/iceberg/pull/16843/changes#diff-a24a1ab4ce975d9b025a9eeb3f7df6f7b4ab24b21534186484ba441eb205a9baR160-R186 handling exceptions thrown during each kafka-client "close()". I believe - CoordinatorThread will absolutely always run "coordinator.stop();" before exiting. Do you happen to see any exceptions thrown with producer.close() [here](https://github.com/apache/iceberg/blob/main/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java#L235-L240) ? Here's why I think join() might be helping for now => KafkaConsumer Thread might call `IcebergSinkTask` -> [close()](https://github.com/apache/iceberg/blob/main/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkTask.java#L63-L65) and follow it with `IcebergSinkTask `-> [open()](https://github.com/apache/iceberg/blob/main/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkTask.java#L58-L60) immediately. (same might happen if the coordinator owning SinkTask restarts) Two cases where open() might get called after close() immediately. 1. Consumer-Partition Rebalance : KafkaConsumer Thread calls SinkTask -> close() and then SinkTask -> open() when partitions are reassigned. 2. Task restart : Kafka Connect Runtime will wait for SinkTask -> stop() before starting another task instance. New task instance immediately joins the consumer-group and results in another SinkTask -> open(). In both cases, we must make sure either 1. existing producer is closed before returning SinkTask -> close() 2. any exception from existing producer (even if thrown later) is handled ([here](https://github.com/apache/iceberg/blob/main/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java#L238-L239)) right now `close()` doesn't block on "coordinator.stop();", so doesn't wait for existing producer to close (via [producer.close()](https://github.com/apache/iceberg/blob/main/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java#L237)). An `open()` immediately after will create a new producer-client. Since these are transactional-producers -> stale producer is expected to throw exception. (even with [close()](https://github.com/a0x8o/kafka/blob/master/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java#L1264-L1280)). So if an open() happens and creates a new producer client - and CoordinatorThread performs `producer.close()` next -> an exception will be thrown and will **skip** 'consumer.close()' (https://github.com/apache/iceberg/blob/main/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java#L238-L239). The current PR might help because it blocks IcebergSinkTask -> close() (https://github.com/apache/iceberg/blob/main/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkTask.java#L67-L83) long enough until CoordinatorThread finishes and closes existing producer (and then consumer) , before next `IcebergSinkTask -> open()` is called. -> But if the join() times-out - we must make sure exceptions from "producer.close()" are handled. I think just handling exceptions [here](https://github.com/apache/iceberg/blob/main/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java#L238-L239) would also prevent leaking consumer. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
