Make Consumer or Producer injectable or make it available for the users

[jonatan-ivanov]

I seems fs2-kafka uses org.apache.kafka.clients.consumer.Consumer and org.apache.kafka.clients.producer.Producer under the hood but users do not have access to them so if they want to use them for integration purposes (pass them to other components), they cannot.

One of the use cases is integration with Micrometer (I'm a maintainer): it is a metrics library with Kafka Client integration and it can make metrics available using org.apache.kafka.clients.consumer.Consumer and org.apache.kafka.clients.producer.Producer.

I'm aware of fs2.kafka.consumer.KafkaMetrics#metrics and even though it's an option to use it in Micrometer, it's less ideal so I would like to explore if making org.apache.kafka.clients.consumer.Consumer and org.apache.kafka.clients.producer.Producer available to the users is an option in fs2-kafka.

[jonatan-ivanov]

@vlovgr What do you think?

[vlovgr]

@jonatan-ivanov We already have MkConsumer and MkProducer for injecting consumers and producers, respectively. Are they enough for what you're trying to achieve?

[sideeffffect]

We'd need something like the opposite -- not injecting, but extracting from an existing fs2-kafka {Producer, Consumer}. So that we can pass the underlying Java Kafka {Producer, Consumer} to

• https://github.com/micrometer-metrics/micrometer/blob/bc2dcb789b12365a2ac3282e8fb1ba99f47e42ed/micrometer-core/src/main/java/io/micrometer/core/instrument/binder/kafka/KafkaClientMetrics.java#L104
• https://github.com/micrometer-metrics/micrometer/blob/bc2dcb789b12365a2ac3282e8fb1ba99f47e42ed/micrometer-core/src/main/java/io/micrometer/core/instrument/binder/kafka/KafkaClientMetrics.java#L163

[vlovgr]

@sideeffffect Since MkConsumer and MkProducer are responsible for supplying the instances, you should also be able to store the instance somewhere. Either by creating it in advance (assuming you also know the settings in advance), or by storing the instance in a Ref at time of creation, and then retrieving it afterwards. Maybe not ideal, but it should work, unless I'm missing something?

[sideeffffect]

Thanks for the response Viktor. Indeed, what you're saying is possible. But as you note, it's very very awkward. I think at this point, a better integration from the side of micrometer would be preferable.

[vlovgr]

Do you know if there are any potential issues with thread safety in this case? At least for the consumer, fs2-kafka uses a dedicated thread for interacting with the Java KafkaConsumer. If not, we could probably consider adding fs2.kafka.consumer.KafkaConsumerClient and fs2.kafka.producer.KafkaProducerClient traits (similar to the existing ones) to expose these. We should still document the general thread safety concerns here though.

[sideeffffect]

adding fs2.kafka.consumer.KafkaConsumerClient and fs2.kafka.producer.KafkaProducerClient traits [...] to expose these.

That would work for me as well. I'd be happy to make a PR to do that or wait for you to implement and publish yourself. Please let me know what you'd prefer.

[vlovgr]

Please go ahead with a pull request.

[jonatan-ivanov]

@jonatan-ivanov We already have MkConsumer and MkProducer for injecting consumers and producers, respectively. Are they enough for what you're trying to achieve?

I think it should be ok but I'm just a messenger here, I opened the issue for @sideeffffect because of micrometer-metrics/micrometer#6267 (I'm a maintainer of Micrometer).