Merge pull request #2019 from zendesk/ben/issue_2016

potential fix for #2016
zendesk · Jul 5, 2023 · 800a489 · 800a489
2 parents 3ab28c6 + c20370c
commit 800a489
Showing 1 changed file with 10 additions and 14 deletions.
diff --git a/src/main/java/com/zendesk/maxwell/producer/AbstractAsyncProducer.java b/src/main/java/com/zendesk/maxwell/producer/AbstractAsyncProducer.java
@@ -30,20 +30,18 @@ public CallbackCompleter(InflightMessageList inflightMessages, Position position
 
 		public void markCompleted() {
 			inflightMessages.freeSlot(messageID);
-			if(isTXCommit) {
-				InflightMessageList.InflightMessage message = inflightMessages.completeMessage(position);
+			InflightMessageList.InflightMessage message = inflightMessages.completeMessage(position);
 
-				if (message != null) {
-					context.setPosition(message.position);
-					long currentTime = System.currentTimeMillis();
-					long endToEndLatency = currentTime - message.eventTimeMS;
+			if (message != null && this.isTXCommit) {
+				context.setPosition(message.position);
+				long currentTime = System.currentTimeMillis();
+				long endToEndLatency = currentTime - message.eventTimeMS;
 
-					messagePublishTimer.update(currentTime - message.sendTimeMS, TimeUnit.MILLISECONDS);
-					messageLatencyTimer.update(Math.max(0L, endToEndLatency - 500L), TimeUnit.MILLISECONDS);
+				messagePublishTimer.update(currentTime - message.sendTimeMS, TimeUnit.MILLISECONDS);
+				messageLatencyTimer.update(Math.max(0L, endToEndLatency - 500L), TimeUnit.MILLISECONDS);
 
-					if (endToEndLatency > metricsAgeSloMs) {
-						messageLatencySloViolationCount.inc();
-					}
+				if (endToEndLatency > metricsAgeSloMs) {
+					messageLatencySloViolationCount.inc();
 				}
 			}
 		}
@@ -84,9 +82,7 @@ public final void push(RowMap r) throws Exception {
 
 		long messageID = inflightMessages.waitForSlot();
 
-		if(r.isTXCommit()) {
-			inflightMessages.addMessage(position, r.getTimestampMillis(), messageID);
-		}
+		inflightMessages.addMessage(position, r.getTimestampMillis(), messageID);
 
 		CallbackCompleter cc = new CallbackCompleter(inflightMessages, position, r.isTXCommit(), context, messageID);