UseCase - 如果来自外部服务的可重试异常并希望从最早的未提交偏移量恢复，则暂停Spring KafkaListener几秒钟。
我遇到的问题 - 以下是实施。
1）没有Seek用法 - 在恢复春天之后kafkalistener正在挑选进入主题分区的最新消息。这是违背目的（缺少最后一次提交的偏移到最新偏移之间的消息）
2）寻求使用 - 我不知道如何获得kafkaconsumer的处理
源代码
消费者中的监听方法
@KafkaListener(topics = "${kafka.consumer.topic}", containerFactory = "kafkaListenerContainerFactory")
public void onReceiving(@Payload ConsumerRecord<String, String> consumerRecord, Acknowledgment acknowledgment) {
try {
Event event = translate(consumerRecord);
someService.processEvent(event, consumerRecord);
commitOffset(acknowledgment)
} catch(ConsumerException e) {
//DO NOT commit offset
}
}
private void commitOffset(Acknowledgment acknowledgment) {
acknowledgment.acknowledge();
}
Service
public void processEvent(Event event, ConsumerRecord<String, String> consumerRecord) {
try {
//call an external API to get realTime event details
//Have a retry on this client
BusinessEntity businessEntity = externalServiceClient.get(event);
//process the Entity
anotherService.process(businessEntity);
} catch(RetryableException re) {
//feign.RetryableException
//we are using feign declarative clients
consumerErrorHandler.handle(re, consumerRecord);
}
}
ErrorHandler - >实现org.springframework.kafka.listener.ErrorHandler
public class ConsumerErrorHandler implements ErrorHandler {
@Autowired
private final KafkaListenerEndpointRegistry registry;
//org.springframework.core.task.SimpleAsyncTaskExecutor
@Autowrired
private final Executor executor;
@Autowired
private Consumer<String, String> kafkaConsumer;
@Override
public void handle(Exception thrownException, ConsumerRecord<?, ?> data) {
//Trying to delegate this to a new Async thread.
executor.execute(() -> {
registry.getListenerContainers().forEach(container -> {
if ((!container.isContainerPaused() || !container.isPauseRequested())) {
log.info("STOPPING_CONSUMER on error");
Optional<TopicPartition> topicPartition = container.getAssignedPartitions().stream().filter(a -> a.partition() == data.partition()).findFirst();
container.pause();
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
log.info("BEFORE_RESUME");
log.info("SEEK CONSUMER before RESUME to this offset: "+data.offset());
topicPartition.ifPresent(a ->
{
log.info("Seek from the current position: " + data.offset());
kafkaConsumer.seek(a, data.offset());
});
container.resume();
log.info("RESUMING_CONSUMER after seek");
topicPartition.ifPresent(a -> {
log.info("CONSUMER is up NOW ??");
});
}
});
});
}
}
消费者配置
private Map<String, Object> consumerConfigs() {
Map<String, Object> confMap = new HashMap<>();
confMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, pubSubServers);
confMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
confMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
confMap.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupIdConfig);
confMap.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "50000");
confMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "50000");
confMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
confMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, OffsetResetStrategy.EARLIEST.name().toLowerCase());
if (this.securityProtocol.equalsIgnoreCase(SSL)) {
confMap.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, this.securityProtocol);
confMap.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,
this.getClass().getResource(clientTrustStoreLocation).getPath());
confMap.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, this.sslTrustStorePassword);
confMap.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG,
this.getClass().getResource(this.clientKeyStoreLocation).getPath());
confMap.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, sslKeyStorePassword);
confMap.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, sslKeyPassword);
confMap.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG,null);
}
return confMap;
}
@Bean
public ConsumerFactory<String, String> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConcurrency("1");
factory.getContainerProperties().setAckOnError(false);
factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
factory.getContainerProperties().setConsumerTaskExecutor(taskExecutor());
factory.setConsumerFactory(consumerFactory());
factory.setErrorHandler(consumerErrorHandler);
factory.setRetryTemplate(retryTemplate());
return factory;
}
@Bean
public AsyncListenableTaskExecutor taskExecutor() {
return createTaskExecutor("1");
}
private RetryTemplate retryTemplate() {
RetryTemplate template = new RetryTemplate();
template.setRetryPolicy(retryPolicy());
template.setBackOffPolicy(backOffPolicy());
return template;
}
private BackOffPolicy backOffPolicy() {
ExponentialBackOffPolicy policy = new ExponentialBackOffPolicy();
policy.setInitialInterval(1000);
return policy;
}
private RetryPolicy retryPolicy() {
SimpleRetryPolicy policy = new SimpleRetryPolicy();
policy.setMaxAttempts("1");
return policy;
}
使用ConsumerAwareErrorHandler。
您无法在另一个线程上执行搜索。请参阅KafkaConsumer javadocs - 它不是线程安全的。
您还必须为其他主题/分区寻找任何剩余记录（除非您只有一个主题/分区）。
最后，在容器暂停之前，你不能退出错误处理程序 - 否则会有一场比赛，消费者可能会在poll()es之前再做另一个pause()。
有关如何执行此类操作的示例，请参阅SeekToCurrentErrorHandler和ContainerStoppingErrorHandler。必须在另一个线程上调用stop()以避免死锁，但是你可以在消费者线程上使用pause()容器（它只是设置一个标志，以便消费者在下一个pause()之前将poll()。
要对容器进行resume()，请使用ApplicationListener或@EventListener来监听暂停容器的容器空闲事件（设置idleEventIterval以获取这些事件。