[axon-kafka] - execption while trying to init token to head

Axon Framework
,
1

Trying to set my init token to head through configurations. Similar code works with axon-connecter.
However getting exception

Configuration

@Configuration
@Profile("kafka-rdbms")
public class KafkaEventProcessorConfig {

	// Value is populated with boolean flag indicating 'past events to be processed'
	// from "application.yaml" file. Default is false.
	@Value("${process.past.events: false}")
	private boolean processPastEvents;

    @Autowired
    public void configure(final EventProcessingConfigurer configurer, StreamableKafkaMessageSource<String, byte[]> streamableKafkaMessageSource) {
        configurer.registerTrackingEventProcessor("info-processor1", c -> streamableKafkaMessageSource);
        configurer.registerTrackingEventProcessor("info-processor2", c -> streamableKafkaMessageSource);
		// if past events are not required to be processed, start TEP from HEAD revision
        if (!processPastEvents)
			configurer.registerTrackingEventProcessorConfiguration(c -> initTokenConfig());
    }
    
	@Bean
	public ListenerInvocationErrorHandler errors() {
		return PropagatingErrorHandler.instance();
	}

	public TrackingEventProcessorConfiguration initTokenConfig() {
		return TrackingEventProcessorConfiguration.forSingleThreadedProcessing()
				.andInitialTrackingToken(streamableKafkaMessageSource -> streamableKafkaMessageSource.createHeadToken());
	}

	@Bean
	public Serializer messageSerializer() {
		return JacksonSerializer.builder().lenientDeserialization().build();
	}
}

Exception

2021-12-01 18:42:32.662  WARN 9920 --- [o-processor1]-0] o.a.e.TrackingEventProcessor             : Fetch Segments for Processor 'info-processor1' failed: null. Preparing for retry in 1s

java.lang.UnsupportedOperationException: null
	at org.axonframework.messaging.StreamableMessageSource.createHeadToken(StreamableMessageSource.java:65) ~[axon-messaging-4.5.3.jar:4.5.3]
	at com.techm.bm.config.KafkaEventProcessorConfig.lambda$3(KafkaEventProcessorConfig.java:47) ~[main/:na]
	at org.axonframework.eventhandling.TrackingEventProcessor$WorkerLauncher.lambda$run$0(TrackingEventProcessor.java:1051) ~[axon-messaging-4.5.3.jar:4.5.3]
	at org.axonframework.common.transaction.TransactionManager.fetchInTransaction(TransactionManager.java:70) ~[axon-messaging-4.5.3.jar:4.5.3]
	at org.axonframework.eventhandling.TrackingEventProcessor$WorkerLauncher.run(TrackingEventProcessor.java:1049) ~[axon-messaging-4.5.3.jar:4.5.3]
	at java.lang.Thread.run(Thread.java:748) [na:1.8.0-262]

2021-12-01 18:42:32.662  WARN 9920 --- [o-processor2]-0] o.a.e.TrackingEventProcessor             : Fetch Segments for Processor 'info-processor2' failed: null. Preparing for retry in 1s

java.lang.UnsupportedOperationException: null
	at org.axonframework.messaging.StreamableMessageSource.createHeadToken(StreamableMessageSource.java:65) ~[axon-messaging-4.5.3.jar:4.5.3]
	at com.techm.bm.config.KafkaEventProcessorConfig.lambda$3(KafkaEventProcessorConfig.java:47) ~[main/:na]
	at org.axonframework.eventhandling.TrackingEventProcessor$WorkerLauncher.lambda$run$0(TrackingEventProcessor.java:1051) ~[axon-messaging-4.5.3.jar:4.5.3]
	at org.axonframework.common.transaction.TransactionManager.fetchInTransaction(TransactionManager.java:70) ~[axon-messaging-4.5.3.jar:4.5.3]
	at org.axonframework.eventhandling.TrackingEventProcessor$WorkerLauncher.run(TrackingEventProcessor.java:1049) ~[axon-messaging-4.5.3.jar:4.5.3]
	at java.lang.Thread.run(Thread.java:748) [na:1.8.0-262]

2021-12-01 18:42:32.677  INFO 9920 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port(s): 6002 (http) with context path ''
2021-12-01 18:42:32.685  INFO 9920 --- [           main] com.techm.bm.Boot                        : Started Boot in 2.258 seconds (JVM running for 39.105)
2021-12-01 18:42:33.663  INFO 9920 --- [o-processor2]-0] o.a.e.TrackingEventProcessor             : Fetching Segments for Processor 'info-processor2' still failing: null. Preparing for retry in 2s
2021-12-01 18:42:33.663  INFO 9920 --- [o-processor1]-0] o.a.e.TrackingEventProcessor             : Fetching Segments for Processor 'info-processor1' still failing: null. Preparing for retry in 2s
2021-12-01 18:42:35.664  INFO 9920 --- [o-processor1]-0] o.a.e.TrackingEventProcessor             : Fetching Segments for Processor 'info-processor1' still failing: null. Preparing for retry in 4s
2021-12-01 18:42:35.664  INFO 9920 --- [o-processor2]-0] o.a.e.TrackingEventProcessor             : Fetching Segments for Processor 'info-processor2' still failing: null. Preparing for retry in 4s
2

That’s to be expected, @Prashant_Shandilya.
That’s why the Kafka Extension throws an UnsupportedOperationException, actually.

Creating a head, tail, or since token with the StreamableKafkaMessageSource is not implemented.
We have an open ticket for this at the moment, which you can find here.

Note that it’s not implemented out of the box because the Kafka Extension requires an entirely different token to be constructed than most sources.
Namely, a Map of TopicPartition to the (in your case head) position of said topic-partition combination.

The easiest step to unblock yourself is to construct the KafkaTrackingToken yourself instead of letting the StreamableKafkaMessageSource do that.
This approach would require you to change your configuration to provide a different initialTrackingToken lambda that does that job for you.