fix: negative Kafka pending

argoproj-labs · Jun 21, 2021 · 24d91a1 · 24d91a1
1 parent 32a73a7
commit 24d91a1
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/runner/sidecar/sources.go b/runner/sidecar/sources.go
@@ -186,7 +186,10 @@ func startKafkaSetPendingLoop(ctx context.Context, x *dfv1.Kafka, sourceName str
 				return
 			}
 			block := rep.GetBlock(x.Topic, partition)
-			totalLags += partitionOffset - block.Offset - 1
+			x := partitionOffset - block.Offset - 1
+			if x > 0 {
+				totalLags += x
+			}
 		}
 		logger.Info("setting pending", "source", sourceName, "pending", totalLags)
 		withLock(func() { step.Status.SourceStatuses.SetPending(sourceName, uint64(totalLags)) })