В моем приложении два канала, которые связаны с двумя темами Kafka:
- input
- error.input.my-group
Вход настроен для отправки сообщения dlq (error.input.my-group) в случае ошибки.
У меня есть StreamListener на «error.input.my-group», который настроен дляотправить сообщение обратно на оригинальный канал.
@StreamListener(Channels.DLQ)
@SendTo(Channels.INPUT)
public Message<?> reRoute(Message<?> failed){
messageDeliveryService.waitUntilCanBeDelivered(failed);
processed.incrementAndGet();
Integer retries = failed.getHeaders().get(X_RETRIES_HEADER, Integer.class);
retries = retries == null ? 1 : retries+1;
if (retries < MAX_RETRIES) {
logger.info("Retry (count={}) for {}", retries, failed);
return buildRetryMessage(failed, retries);
}
else {
logger.error("Retries exhausted (-> sent to parking lot) for {}", failed);
Channels.parkingLot().send(MessageBuilder.fromMessage(failed)
.setHeader(BinderHeaders.PARTITION_OVERRIDE,
failed.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID))
.build());
}
return null;
}
private Message<?> buildRetryMessage(Message<?> failed, int retries) {
return MessageBuilder.fromMessage(failed)
.setHeader(X_RETRIES_HEADER, retries)
.setHeader(BinderHeaders.PARTITION_OVERRIDE,
failed.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID))
.build();
}
Вот мой класс каналов
@Component
public interface Channels {
String INPUT = "INPUT";
//Default name use by SCS (error.<input-topic-name>.<group-name>)
String DLQ = "error.input.my-group";
String PARKING_LOT = "parkingLot.input.my-group";
@Input(INPUT)
SubscribableChannel input();
@Input(DLQ)
SubscribableChannel dlq();
@Output(PARKING_LOT)
MessageChannel parkingLot();
}
Вот моя конфигурация
spring:
cloud:
stream:
default:
group: my-group
binder:
headerMode: headers kafka:
binder:
# Necessary in order to commit the message to all the Kafka brokers handling the partition -> maximum durability
# -1 = all
requiredAcks: -1
brokers: bootstrap.kafka.svc.cluster.local:9092,bootstrap.kafka.svc.cluster.local:9093,bootstrap.kafka.svc.cluster.local:9094,bootstrap.kafka.svc.cluster.local:9095,bootstrap.kafka.svc.cluster.local:9096,bootstrap.kafka.svc.cluster.local:9097
bindings:
input:
consumer:
partitioned: true
enableDlq: true
dlqProducerProperties:
configuration:
key.serializer: "org.apache.kafka.common.serialization.ByteArraySerializer"
"[error.input.my-group]":
consumer:
# We cannot loose any message and we don't have any DLQ for the DLQ, therefore we only commit in case of success
autoCommitOnError: false
ackEachRecord: true
partitioned: true
enableDlq: false
bindings:
input:
contentType: application/xml
destination: input
"[error.input.my-group]":
contentType: application/xml
destination: error.input.my-group
"[parkingLot.input.my-group]":
contentType: application/xml
destination: parkingLot.input.my-group
Проблема в том, что мои сообщения никогда не бываютснова нажал на Кафку, но напрямую доставил на мой входной каналЧто-то я неправильно понял?