Provide an extension point for setting offsets before starting container

@dsyer commented on Mon Sep 09 2019

Currently there seems to be no way to expose the equivalent of this using Spring Cloud Stream:

@KafkaListener(topicPartitions = { @TopicPartition(partitionOffsets = {
			@PartitionOffset(partition = "#{config.partition}", initialOffset = "#{config.offset}") }, topic = "#{config.topic}") })
public void consumer(Message<?> message) {
	...
}

The only flexibility with offsets in Spring Cloud Stream is to set the consumer to seek to "BEGINNING" or "END" (nothing that corresponds to an actual long offset).

This will be essential for providing the "effectively once" semantics for quality of service (storing offsets outside Kafka).

@garyrussell commented on Mon Sep 09 2019

Add a ListenerContainerCustomizer. container.getContainerProperties().setConsumerRebalanceListener(...).

Use a ConsumerAwareRebalanceListener and consumer.seek(...)s.

Sent from 'phone... apologies for lack of verbosity and formatting.

@garyrussell commented on Mon Sep 09 2019

Oh. That will only work with group management. Your question uses manual partition assignment. There is no hook for setting offsets in that mode.

@dsyer commented on Tue Sep 10 2019

Actually it does seem to work - I don't need manual partition assignment; it was just the only way to make @KafkaListener accept an initial offset. Here's what I have:

@SpringBootApplication
public class DemoApplication {

	public static void main(String[] args) throws Exception {
		new SpringApplicationBuilder(DemoApplication.class).run(args);
	}

	@Bean
	public Consumer<Message<?>> consumer() {
		return message -> {
			System.err.println(message);
		};
	}

	@Bean
	public ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> listenerCustomizer() {
		return new ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>>() {

			@Override
			public void configure(AbstractMessageListenerContainer<?, ?> container,
					String destinationName, String group) {
				System.err.println("Customizing: " + group);
				container.getContainerProperties().setConsumerRebalanceListener(
						new ConsumerAwareRebalanceListener() {
							@Override
							public void onPartitionsAssigned(
									org.apache.kafka.clients.consumer.Consumer<?, ?> consumer,
									Collection<TopicPartition> partitions) {
								for (TopicPartition partition : partitions) {
									System.err.println("Seeking: " + partition);
									consumer.seek(partition, 3);
								}
							}
						});
			}
		};
	}

}

I only see messages from offset 3, independent of whether I set spring.cloud.stream.bindings.input.group or not. This is perfect if it is actually working as it appears to be. We could document it a bit, I guess.

@dsyer commented on Tue Sep 10 2019

Also, if this is the only way to do it, there is another problem which is that there is only one ListenerContainerCustomizer and only one ConsumerRebalanceListener, so if I make a library with one of these, there is no room for additional customization by the user.

Start with ListenerContainerCustomizer, AbstractMessageListenerContainer, and ConsumerAwareRebalanceListener to understand the existing customization path and offset seeking behavior. Define how multiple customizers or rebalance listeners can coexist, then verify that a user can configure the required offset behavior without blocking library-provided customization.