pjfanning opened a new pull request, #650: URL: https://github.com/apache/pekko-connectors-kafka/pull/650
### Motivation Two paths leak Kafka clients. `DeferredProducer.resolveProducer` assigns the asynchronously created producer through `AsyncCallback.invoke`, which is silently dropped once the stage has completed (`GraphInterpreter.runAsyncInput` discards the event when the stage is completed and only reports it through `invokeWithFeedback`). When the stage fails or is cancelled while the producer is still being created, `assignProducer` never runs, so `producer` stays `null`, the lifecycle stays `AsyncCreateRequestSent`, and `closeProducer()` — gated on `producerAssignmentLifecycle == Assigned` — does nothing. The fully constructed `KafkaProducer` (sender thread, sockets, buffer pool) is then orphaned for the lifetime of the JVM. This is not a narrow race: `createKafkaProducerAsync` is `enriched.map(createKafkaProducer)`, and mapping over an already completed future with the materializer execution context still schedules asynchronously, so `producerFuture.value` is normally `None` even without async enrichment. It affects `DefaultProducerStage`, `TransactionalProducerStage` and `CommittingProducerSinkStage`. `KafkaConsumerActor.postStop` calls the user supplied partition assignment handler before closing the consumer: - a handler that throws in `onStop` skips `consumer.close(...)`, so the consumer and its consumer group membership are only released when the session times out, and the client itself is never released; - the same line throws a `NullPointerException` when the actor is stopped before the settings have been applied (`Stop` handled in `expectSettings`, or a failed settings enrichment), because `consumer` and `settings` are still `null`. That NPE replaces the real stop reason in the logs. ### Modification - `DeferredProducer`: assign the producer with `invokeWithFeedback` and close the producer when the callback is rejected because the stage is already gone. The close path runs outside the stage logic, so it only reads the immutable settings. - `KafkaConsumerActor.postStop`: wrap the partition assignment handler call in `try`/`finally`, log a handler failure instead of letting it abort the stop, and only close the consumer when one was created. ### Result Producers created after their stage completed are closed instead of orphaned, and the consumer is closed even when the partition assignment handler fails or the actor is stopped before it has a consumer. ### Tests - `sbt "tests/Test/testOnly org.apache.pekko.kafka.internal.*"` — 137 succeeded, 0 failed. The three new tests (one in `ProducerSpec`, two in `ConsumerSpec`) fail without the fixes: the producer mock is never closed, and the pre-initialisation stop logs a `NullPointerException`. - `sbt core/mimaReportBinaryIssues` — success - `sbt scalafmtCheckAll headerCheckAll` — success ### References None - found while auditing the main sources for resource leaks -- 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]
