我已经将Spring Kafka监听器初始化为
@Bean
public Map<String, Object> consumerConfig() {
final HashMap<String, Object> result = new HashMap<>();
result.put(BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
result.put(GROUP_ID_CONFIG, groupId);
result.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
result.put(VALUE_DESERIALIZER_CLASS_CONFIG, MyKafkaJacksonRulesExecutionResultsDeserializer.class);
return result;
}
@Bean
public ConsumerFactory<Long, MessageResult> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfig());
}
@Bean
public ConcurrentKafkaListenerContainerFactory<Long, MessageResult> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Long, MessageResult> containerFactory = new ConcurrentKafkaListenerContainerFactory<>();
containerFactory.setConsumerFactory(consumerFactory());
containerFactory.setConcurrency(KAFKA_LISTENER_THREADS_COUNT);
containerFactory.getContainerProperties().setPollTimeout(KAFKA_LISTENER_POLL_TIMEOUT);
containerFactory.getContainerProperties().setAckOnError(true);
containerFactory.getContainerProperties().setAckMode(RECORD);
return containerFactory;
}
并用作
@KafkaListener(topics = "${spring.kafka.out-topic}")
public void processSrpResults(MessageResult result) {
反序列化器在反序列化过程中引发异常,这会导致无限循环,因为监听器无法获取消息。
我怎样才能使kafka监听器在错误时提交?
最佳答案
我创建了反序列化器的子类,该子类引发了异常。然后,在我的配置中将其用作解串器。然后,您的处理器必须处理空对象。
public class MyErrorHandlingDeserializer extends ExceptionThrowingDeserializer {
@Override
public Object deserialize(String topic, byte[] data) {
try {
return super.deserialize(topic, data);
} catch (Exception e) {
log.error("Problem deserializing data " + new String(data) + " on topic " + topic, e);
return null;
}
}
}
关于Spring Kafka监听器无限循环出错,我们在Stack Overflow上找到一个类似的问题:https://stackoverflow.com/questions/44796543/