Provide a way to create an OffsetAndMetdata with metadata when using Spring-Kafka

[frosiere]

In some specific cases, it may make sense to set/customize the metadata field of the OffsetAndMetadata object with data coming from the consumer. When using Spring Kafka, it's impossible as this object is created internally in KafkaMessageListenerContainer.ListenerConsumer.

Due to this, a callback can be set on the container properties but the metadata will always be empty when calling the onComplete method of the callback.

This may help having a mechanism to set these metadata.

So, the proposal would be to either review the way to set a callback or to simply have an OffsetAndMetadataProvider as follow

public interface OffsetAndMetadataProvider {

   OffsetAndMetadata provide(Consumer<?, ?> consumer, long offset);
}

This provider would then be set on the container properties as show in the following portion of code

public class MyOffsetAndMetadataProvider  implements OffsetAndMetadataProvider {

   @Override   
   OffsetAndMetadata provide(Consumer<?, ?> consumer, long offset) {
      // in this case, metadata would be the consumer group id...
      return new OffsetAndMetadata(offset, consumer.groupMetadata().groupId());
   }
}

// at consumer factory level
...
getContainerProperties().setOffsetAndMetadataProvider(new MyOffsetAndMetadataProvider());
...

Internally, Spring Kafka would need to be slightly reworked to use the following kind of method instead of directly creating the OffsetAndMetadata object.

// in KafkaMessageListenerContainer.ListenerConsumer
// would be called 6 times
private OffsetAndMetadata createOffsetAndMetadata(long offset) {
   final OffsetAndMetadataProvider offsetAndMetadataProvider = containerProperties.getOffsetAndMetadataProvider();
   return offsetAndMetadataProvider == null
         ? new OffsetAndMetadata(offset)
         : offsetAndMetadataProvider.provide(consumer, offset)
}

Any other way to set metadata is more than welcome.

If we agree on the need, I can contribute on the implementation. This one or another one.

Thanks for your reply.

[garyrussell]

Thanks; the example implementation makes it clearer than the SO question.

However, I am not sure we want to provide full access to the consumer, just to get the group id.

Perhaps we can add some class with useful information...

public interface OffsetAndMetadataProvider extends BiFunction<OffsetAndMetadata, ListeneMetadata, OffsetAndMetadata> {

   OffsetAndMetadata apply(OffsetAndMetadata oam, ListenerMetadata lm);

}

public class ListenerMetadata {

    String groupId;
 
    String listenerId;

    ...

}

We can convert this to a new feature when the discussion is complete; a contribution would be appreciated.

[garyrussell]

I deleted the comment about the thread local - the callback is called on a different thread.

[frosiere]

Fine for me, I like the idea of exposing the listener metadata instead of the consumer.
Any advantage of using a BiFunction for the provider in this particular case?

To have a stronger link and to avoid mistakes between the callback and the provider, we may expose a new setter on the ContainerProperties overloading the existing one
setCommitCallback(OffsetCommitCallback commitCallback, OffsetAndMetadataProvider provider)

We also need to define the properties we want to expose as listener metadata, then, I can create a ticket and make a merge request.

[garyrussell]

Agreed, no real need to extend BiConsumer<?, ?, ?>

I meant to add an elipsis to imply there might be more interesting properties`; I suggest you submit a PR with your initial thoughts and we can continue the discussion there.

Creating an issue from this.

[garyrussell]

#2170

[frosiere]

Discussion can be closed due to #2170