pjfanning opened a new pull request, #621:
URL: https://github.com/apache/pekko-projection/pull/621
### Motivation
`KafkaSourceProviderImpl.source` creates the `MetadataClientAdapter` - and
with it a `KafkaConsumerActor` and its Kafka consumer - before returning the
source:
```scala
val metadataClient = metadataClientFactory()
val numPartitionsF = metadataClient.numPartitions(topics)
numPartitionsF.failed.foreach(_ => metadataClient.stop())
numPartitionsF.map { numPartitions =>
_source(readOffsets, numPartitions, metadataClient)
...
.mapMaterializedValue { terminated => terminated.onComplete(_ =>
metadataClient.stop()); NotUsed }
}
```
It is stopped only when `numPartitions` fails, or when the materialized
stream terminates. If the returned source is never materialized - for example
when the projection is stopped between `source()` completing and the stream
being run, which the restart backoff makes possible on every restart - the
client and its consumer actor are never stopped.
### Modification
Create the metadata client inside `Source.lazyFutureSource`, so it is only
created when the source is materialized and demand arrives, and is then always
stopped by the existing `watchTermination` callback. The `numPartitions` lookup
(used for the `flatMapMerge` breadth) moves with it, so a failure now fails the
stream instead of the returned future; both end up failing the projection
stream, which the restart backoff handles.
### Result
No orphaned metadata client, and no orphaned Kafka consumer actor, for a
source that is never materialized.
### Tests
- New `KafkaSourceProviderImplSpec` case: asserts no metadata client is
created for a source that is not materialized, and that one is created and then
stopped when the stream is run. It fails before the fix (the client is created
eagerly).
- `sbt "kafka/test"` - 2 passed
- `sbt "kafka/mimaReportBinaryIssues"` - success
- `sbt checkCodeStyle` - success
- The `kafka-test` integration specs were not run locally (they need Kafka
testcontainers); they exercise this path via
`Source.futureSource(provider.source(...))` and are covered by the CI Kafka
integration jobs.
### References
None - found during a resource-leak audit of the main sources
--
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]
