Kafka error handling: Decide the expectation and action

ConcurrentKafkaListenerContainerFactory<SpecificRecord, SpecificRecord> factory =new ConcurrentKafkaListenerContainerFactory<>();// Deciding whether to commit offset for every record or whole batch.factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.BATCH);factory.setBatchListener(true); // By default is false.// How may listeners to create to have parallelism in processing multiple batch together.factory.setConcurrency(Integer.parseInt(concurrency));// As batch listener is true so error handler also needs to be set for batch.factory.setBatchErrorHandler(errorHandler);// Defining properties for consumers.DefaultKafkaConsumerFactory<SpecificRecord, SpecificRecord> consumerFactory = getKafkaConsumerFactory();factory.setConsumerFactory(consumerFactory);private DefaultKafkaConsumerFactory<SpecificRecord, SpecificRecord> getKafkaConsumerFactory() {Map<String, Object> props = kafkaProperties.buildConsumerProperties();props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, SpecificAvroDeserializer.class);props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, SpecificAvroDeserializer.class);props.put("schema.registry.url", kafkaProperties.getProperties().get("schema.registry.url"));return new DefaultKafkaConsumerFactory<>(props);}
@Beanpublic ConsumerRecordRecoverer recordRecoverer(ApplicationContext context) {return (consumerRecord, exception) -> {//Define action};}

--

--

Get the Medium app

A button that says 'Download on the App Store', and if clicked it will lead you to the iOS App store
A button that says 'Get it on, Google Play', and if clicked it will lead you to the Google Play store