fix: not considered as back pressured when onFull is discardLatest (#…

…1153) Signed-off-by: Derek Wang <[email protected]>
numaproj · Oct 4, 2023 · 00e6b6a · 00e6b6a
1 parent b271910
commit 00e6b6a
Showing 1 changed file with 4 additions and 0 deletions.
diff --git a/pkg/reconciler/vertex/scaling/scaling.go b/pkg/reconciler/vertex/scaling/scaling.go
@@ -456,6 +456,10 @@ func (s *Scaler) hasBackPressure(pl dfv1.Pipeline, vertex dfv1.Vertex) (bool, bo
 	directPressure, downstreamPressure := false, false
 loop:
 	for _, e := range downstreamEdges {
+		if e.BufferFullWritingStrategy() == dfv1.DiscardLatest {
+			// If the edge is configured to discard latest on full, we don't consider it as back pressure.
+			continue
+		}
 		vertexKey := pl.Namespace + "/" + pl.Name + "-" + e.To
 		pendingVal, ok := s.vertexMetricsCache.Get(vertexKey + "/pending")
 		if !ok { // Vertex key has not been cached, skip it.