SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1046: Fix Out of Order Offset Commit with DLQ

Browse Source 
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1046

When using ack mode `MANUAL_IMMEDIATE`, we must acknowledge the delivery on
the container thread, otherwise the commit will be queued for later processing,
possibly causing out of order commits, incorrectly reducing the committed offset.

Wait for the send to complete before acknowledging; use a timeout slightly larger
than the configured producer delivery timeout to avoid premature timeouts.

Tested with user-provided test case.

Fix missing timeout buffer.

More context for this issue: https://gitter.im/spring-cloud/spring-cloud-stream?at=6050f43595e23446e43cacd1
This commit is contained in:
Gary Russell
2021-03-19 10:16:33 -04:00
committed by Soby Chacko
parent a7e025794c
commit f8b290844b
1 changed files with 34 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -30,6 +30,7 @@ import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;
import java.util.regex.Pattern;
@@ -1141,8 +1142,21 @@ public class KafkaMessageChannelBinder extends
final KafkaTemplate<?, ?> kafkaTemplate = new KafkaTemplate<>(
producerFactory);
Object timeout = producerFactory.getConfigurationProperties().get(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG);
Long sendTimeout = null;
if (timeout instanceof Number) {
sendTimeout = ((Number) timeout).longValue() + 2000L;
}
else if (timeout instanceof String) {
sendTimeout = Long.parseLong((String) timeout) + 2000L;
}
if (timeout == null) {
sendTimeout = ((Integer) ProducerConfig.configDef()
.defaultValues()
.get(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG)).longValue() + 2000L;
}
@SuppressWarnings("rawtypes")
DlqSender<?, ?> dlqSender = new DlqSender(kafkaTemplate);
DlqSender<?, ?> dlqSender = new DlqSender(kafkaTemplate, sendTimeout);
return (message) -> {
@@ -1574,8 +1588,11 @@ public class KafkaMessageChannelBinder extends
private final KafkaTemplate<K, V> kafkaTemplate;
DlqSender(KafkaTemplate<K, V> kafkaTemplate) {
private final long sendTimeout;
DlqSender(KafkaTemplate<K, V> kafkaTemplate, long timeout) {
this.kafkaTemplate = kafkaTemplate;
this.sendTimeout = timeout;
}
@SuppressWarnings("unchecked")
@@ -1609,18 +1626,26 @@ public class KafkaMessageChannelBinder extends
public void onSuccess(SendResult<K, V> result) {
if (KafkaMessageChannelBinder.this.logger.isDebugEnabled()) {
KafkaMessageChannelBinder.this.logger
.debug("Sent to DLQ " + sb.toString());
}
if (ackMode == ContainerProperties.AckMode.MANUAL || ackMode == ContainerProperties.AckMode.MANUAL_IMMEDIATE) {
messageHeaders.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class).acknowledge();
.debug("Sent to DLQ " + sb.toString() + ": " + result.getRecordMetadata());
}
}
});
try {
sentDlq.get(this.sendTimeout, TimeUnit.MILLISECONDS);
}
catch (InterruptedException ex) {
Thread.currentThread().interrupt();
throw ex;
}
}
catch (Exception ex) {
if (sentDlq == null) {
KafkaMessageChannelBinder.this.logger
.error("Error sending to DLQ " + sb.toString(), ex);
KafkaMessageChannelBinder.this.logger
.error("Error sending to DLQ " + sb.toString(), ex);
}
finally {
if (ackMode == ContainerProperties.AckMode.MANUAL
|| ackMode == ContainerProperties.AckMode.MANUAL_IMMEDIATE) {
messageHeaders.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class).acknowledge();
}
}