I had created a Kafka consumer using Azure Event Hub and Spring Boot, but when I start the application and then begin to try to consume the messages that I have in my broker I receive the error bellow
java.lang.IllegalStateException: This error handler cannot process 'org.apache.kafka.common.errors.InvalidSessionTimeoutException's; no record information is available
at org.springframework.kafka.listener.SeekUtils.seekOrRecover(SeekUtils.java:151) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
at org.springframework.kafka.listener.SeekToCurrentErrorHandler.handle(SeekToCurrentErrorHandler.java:103) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.handleConsumerException(KafkaMessageListenerContainer.java:1247) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1004) ~[spring-kafka-2.5.2.RELEASE.jar:2.5.2.RELEASE]
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[na:na]
at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264) ~[na:na]
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java) ~[na:na]
at java.base/java.lang.Thread.run(Thread.java:832) ~[na:na]
Caused by: org.apache.kafka.common.errors.InvalidSessionTimeoutException: The session timeout is not within the range allowed by the broker (as configured by group.min.session.timeout.ms and group.max.session.timeout.ms).
2020-07-22 17:09:28.017 INFO 2582630 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=ed3c3738-5b4d-4a83-8c19-76f0a11316bb-0, groupId=com.hi.servicemessagestatusprocessor.ServiceMessageStatusProcessorApplication] (Re-)joining group
2020-07-22 17:09:28.035 ERROR 2582630 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=ed3c3738-5b4d-4a83-8c19-76f0a11316bb-0, groupId=com.hi.servicemessagestatusprocessor.ServiceMessageStatusProcessorApplication] Attempt to join group failed due to fatal error: The session timeout is not within the range allowed by the broker (as configured by group.min.session.timeout.ms and group.max.session.timeout.ms).
2020-07-22 17:09:28.035 INFO 2582630 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=ed3c3738-5b4d-4a83-8c19-76f0a11316bb-0, groupId=com.hi.servicemessagestatusprocessor.ServiceMessageStatusProcessorApplication] Join group failed with org.apache.kafka.common.errors.InvalidSessionTimeoutException: The session timeout is not within the range allowed by the broker (as configured by group.min.session.timeout.ms and group.max.session.timeout.ms).
2020-07-22 17:09:28.036 ERROR 2582630 --- [ntainer#0-0-C-1] essageListenerContainer$ListenerConsumer : Consumer exception
I can't figure out how to resolve this error, bellow have my configuration to Kafka consumer
@EnableKafka
@Configuration
public class KafkaConsumerConfig {
public ConsumerFactory<String, HilabMessage> consumerFactory(String groupId) throws IOException {
InputStream kafkaProperties = getClass().getClassLoader()
.getResourceAsStream("kafka.properties");
Properties properties = new Properties();
properties.load(kafkaProperties);
Map<String, Object> configs = new HashMap<>();
for (final String name: properties.stringPropertyNames())
configs.put(name, properties.getProperty(name));
configs.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
configs.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "1");
configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
configs.put(ConsumerConfig.CLIENT_ID_CONFIG, UUID.randomUUID().toString());
configs.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class.getName());
return new DefaultKafkaConsumerFactory<>(
configs, new StringDeserializer(), new JsonDeserializer<>(HilabMessage.class));
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, HilabMessage> kafkaListenerContainerFactory() throws IOException {
ConcurrentKafkaListenerContainerFactory<String, HilabMessage> factory = new
ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory(ServiceMessageStatusProcessorApplication.class.getName()));
return factory;
}
}
Note that my producer has a similar configuration and it publishes the messages to the broker