fix(sidecar): fix Kafka pending tiny over-estimate

argoproj-labs · Jun 17, 2021 · 2b90f11 · 2b90f11
1 parent 1f29603
commit 2b90f11
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 2 deletions.
diff --git a/Procfile b/Procfile
@@ -2,4 +2,4 @@ controller: ARGO_DATAFLOW_PULL_POLICY=IfNotPresent ARGO_DATAFLOW_UPDATE_INTERVAL
 logs: make logs
 argocli: make argocli
 ui: make ui
-test: kubectl port-forward svc/testapi 8378:8378
+testapi: until kubectl port-forward svc/testapi 8378:8378; do sleep 3s; done
diff --git a/runner/sidecar/sources.go b/runner/sidecar/sources.go
@@ -186,7 +186,7 @@ func startKafkaSetPendingLoop(ctx context.Context, x *dfv1.Kafka, sourceName str
 				return
 			}
 			block := rep.GetBlock(x.Topic, partition)
-			totalLags += partitionOffset - block.Offset
+			totalLags += partitionOffset - block.Offset - 1
 		}
 		logger.Info("setting pending", "source", sourceName, "pending", totalLags)
 		withLock(func() { step.Status.SourceStatuses.SetPending(sourceName, uint64(totalLags)) })