Revert "potential fix for #2016"

Right idea, but dumb implementation. This reverts commit c20370c.
zendesk · Sep 9, 2023 · 1849ee4 · 1849ee4
1 parent 21b104e
commit 1849ee4
Showing 1 changed file with 14 additions and 10 deletions.
diff --git a/src/main/java/com/zendesk/maxwell/producer/AbstractAsyncProducer.java b/src/main/java/com/zendesk/maxwell/producer/AbstractAsyncProducer.java
@@ -30,18 +30,20 @@ public CallbackCompleter(InflightMessageList inflightMessages, Position position
 
 		public void markCompleted() {
 			inflightMessages.freeSlot(messageID);
-			InflightMessageList.InflightMessage message = inflightMessages.completeMessage(position);
+			if(isTXCommit) {
+				InflightMessageList.InflightMessage message = inflightMessages.completeMessage(position);
 
-			if (message != null && this.isTXCommit) {
-				context.setPosition(message.position);
-				long currentTime = System.currentTimeMillis();
-				long endToEndLatency = currentTime - message.eventTimeMS;
+				if (message != null) {
+					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();
+					}
 				}
 			}
 		}
@@ -81,7 +83,9 @@ public final void push(RowMap r) throws Exception {
 
 		long messageID = inflightMessages.waitForSlot();
 
-		inflightMessages.addMessage(position, r.getTimestampMillis(), messageID);
+		if(r.isTXCommit()) {
+			inflightMessages.addMessage(position, r.getTimestampMillis(), messageID);
+		}
 
 		CallbackCompleter cc = new CallbackCompleter(inflightMessages, position, r.isTXCommit(), context, messageID);