github-actions[bot] commented on code in PR #41075:
URL: https://github.com/apache/beam/pull/41075#discussion_r4504896829
##########
sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:
##########
@@ -2069,126 +2159,138 @@ func (ss *stageState) timeSortedDataReadyLocked(ts,
watermark mtime.Time) bool {
// buildEventTimeBundle for aggregation stages, processes all elements that
are within the watermark for completed windows.
func (*aggregateStageKind) buildEventTimeBundle(ss *stageState, watermark
mtime.Time) (toProcess elementHeap, _ mtime.Time, _ set[string], _
map[mtime.Time]int, panesInBundle []bundlePane, schedulable bool,
pendingAdjustment int) {
minTs := mtime.MaxTimestamp
- // TODO: Allow configurable limit of keys per bundle, and elements per
key to improve parallelism.
- // TODO: when we do, we need to ensure that the stage remains
schedualable for bundle execution, for remaining pending elements and keys.
- // With the greedy approach, we don't need to since "new data" triggers
a refresh, and so should completing processing of a bundle.
newKeys := set[string]{}
-
holdsInBundle := map[mtime.Time]int{}
-
accumulatingPendingAdjustment := 0
+
+ candidateKeys := set[string]{}
+ for k := range ss.pendingByKeys {
+ candidateKeys.insert(k)
+ }
+ for k := range ss.pendingOnTime {
+ candidateKeys.insert(k)
+ }
+
keysPerBundle:
- for k, dnt := range ss.pendingByKeys {
+ for k := range candidateKeys {
if ss.inprogressKeys.present(k) {
continue
}
- newKeys.insert(k)
- // Track the min-timestamp for later watermark handling.
- if dnt.elements[0].timestamp < minTs {
- minTs = dnt.elements[0].timestamp
- }
- var toProcessForKey []element
- var toSkipForKey []element
- // Can we pre-compute this bit when adding to pendingByKeys?
- // startBundle is in run in a single scheduling goroutine, so
moving per-element code
- // to be computed by the bundle parallel goroutines will speed
things up a touch.
- for dnt.elements.Len() > 0 {
- // For aggregations, only include elements for this key
- // if we're after the end of window, or after the
window expiry deadline.
- // We will only ever trigger aggregations by watermark
at most twice, once the watermark passes the window ends for OnTime completion,
- // and once for when the window is closing.
- elm := heap.Pop(&dnt.elements).(element)
- shouldSkip := false
-
- if watermark <= elm.window.MaxTimestamp() {
- // The watermark hasn't passed the end of the
window yet, we do nothing.
- shouldSkip = true
- }
- // Watermark is past the end of this window. Have we
fired an OnTime pane yet?
- state :=
ss.state[LinkID{}][elm.window][string(elm.keyBytes)]
- // If this is not the ontime firing for this key.
-
- if !shouldSkip && state.Pane.Timing != typex.PaneEarly
&& watermark <= ss.strat.EarliestCompletion(elm.window) {
- // The watermark is still before the earliest
final completion for this window.
- // Do not add further data for this firing.
- // If this is the Never trigger, we also don't
fire OnTime until after the earliest completion.
- shouldSkip = true
- }
- if !shouldSkip && ss.strat.IsNeverTrigger() &&
watermark <= ss.strat.EarliestCompletion(elm.window) {
- // The NeverTrigger only has a single firing at
the end of window + allowed lateness.
- shouldSkip = true
+ var firingWindows []typex.Window
+ firingSet := set[typex.Window]{}
+ var newByWin map[typex.Window][]element
+
+ if dnt := ss.pendingByKeys[k]; dnt != nil {
+ var toSkipForKey []element
+ for dnt.elements.Len() > 0 {
+ // Aggregate by watermark at most twice per
window: once at window end
+ // for OnTime, and once at window closing after
allowed lateness.
+ elm := heap.Pop(&dnt.elements).(element)
+ shouldSkip := false
+
+ if watermark <= elm.window.MaxTimestamp() {
+ shouldSkip = true
+ }
+ state := ss.state[LinkID{}][elm.window][k]
+ onTimePassed := state.Pane.Timing !=
typex.PaneEarly || elm.window.MaxTimestamp() < ss.output
+ if !shouldSkip && onTimePassed && watermark <=
ss.strat.EarliestCompletion(elm.window) {
+ shouldSkip = true
+ }
+ if !shouldSkip && ss.strat.IsNeverTrigger() &&
watermark <= ss.strat.EarliestCompletion(elm.window) {
+ shouldSkip = true
+ }
+
+ if shouldSkip {
+ toSkipForKey = append(toSkipForKey, elm)
+ continue
+ }
+ if newByWin == nil {
+ newByWin = map[typex.Window][]element{}
+ }
+ newByWin[elm.window] =
append(newByWin[elm.window], elm)
+ if !firingSet.present(elm.window) {
+ firingSet.insert(elm.window)
+ firingWindows = append(firingWindows,
elm.window)
+ }
}
- if shouldSkip {
- toSkipForKey = append(toSkipForKey, elm)
- } else {
- toProcessForKey = append(toProcessForKey, elm)
+ for _, elm := range toSkipForKey {
+ heap.Push(&dnt.elements, elm)
+ }
+ if dnt.elements.Len() == 0 {
+ delete(ss.pendingByKeys, k)
}
}
- // Re-add skipped elements to the heap.
- // This ensures that `dnt.elements` reflects only the truly
skipped elements for future processing.
- for _, elm := range toSkipForKey {
- heap.Push(&dnt.elements, elm)
+ for win := range ss.pendingOnTime[k] {
+ if watermark <= win.MaxTimestamp() {
+ continue
+ }
+ if ss.strat.IsNeverTrigger() && watermark <=
ss.strat.EarliestCompletion(win) {
+ continue
+ }
+ if !firingSet.present(win) {
+ firingSet.insert(win)
+ firingWindows = append(firingWindows, win)
+ }
}
- // Get the pane for the aggregation correct, only mutate it
once per key and window.
- handledWindows := set[typex.Window]{}
- for _, elm := range toProcessForKey {
- state :=
ss.state[LinkID{}][elm.window][string(elm.keyBytes)]
- if handledWindows.present(elm.window) {
- // The pane is already correct for this key +
window + firing.
- if ss.strat.Accumulating && !state.Pane.IsLast {
- // If this isn't the last pane, then we
must add the element back to the pending store for subsequent firings.
- heap.Push(&dnt.elements, elm)
- accumulatingPendingAdjustment++
- }
+ var toProcessForKey []element
+ for _, win := range firingWindows {
+ ss.initAggStateLocked(win, k)
+ state := ss.state[LinkID{}][win][k]
+ removed := ss.takePendingOnTimeLocked(win, k,
holdsInBundle)
+ newElems := newByWin[win]
+ priorCount := len(state.Accumulated)
+ if priorCount+len(newElems) == 0 {
continue
}
- handledWindows.insert(elm.window)
- state.Pane = computeNextWatermarkPane(state.Pane)
- // Determine if this is the last pane.
- // Check if this is the post closing firing, which will
be the last one.
- // Unless it's the ontime pane, at which point it can
never be last.
- if watermark > ss.strat.EarliestCompletion(elm.window)
&& state.Pane.Timing != typex.PaneOnTime {
+ winElems := make([]element, 0, priorCount+len(newElems))
+ winElems = append(winElems, state.Accumulated...)
+ winElems = append(winElems, newElems...)
+
+ state.Pane = computeNextWatermarkPane(state.Pane,
win.MaxTimestamp() < ss.output)
+ if watermark > ss.strat.EarliestCompletion(win) &&
state.Pane.Timing != typex.PaneOnTime {
state.Pane.IsLast = true
}
if ss.strat.AllowedLateness == 0 ||
ss.strat.IsNeverTrigger() {
- // If the allowed lateness is zero, then this
will be the last pane.
- // If this is the NeverTrigger, it's the last
pane.
state.Pane.IsLast = true
}
- ss.state[LinkID{}][elm.window][string(elm.keyBytes)] =
state
+ if ss.strat.Accumulating && !state.Pane.IsLast &&
watermark <= ss.strat.EarliestCompletion(win) {
+ nextAcc := make([]element, len(winElems))
+ copy(nextAcc, winElems)
+ state.Accumulated = nextAcc
+ } else {
+ state.Accumulated = nil
+ }
+ ss.state[LinkID{}][win][k] = state
- // Save latest PaneInfo for this window + key pair. It
will be used in PersistBundle.
panesInBundle = append(panesInBundle, bundlePane{
- win: elm.window,
- key: string(elm.keyBytes),
- pane:
ss.state[LinkID{}][elm.window][string(elm.keyBytes)].Pane,
+ win: win,
+ key: k,
+ pane: state.Pane,
})
-
- // The pane is already correct for this key + window +
firing.
- if ss.strat.Accumulating && !state.Pane.IsLast {
- // If this isn't the last pane, then we must
add the element back to the pending store for subsequent firings.
- heap.Push(&dnt.elements, elm)
- accumulatingPendingAdjustment++
- }
+ accumulatingPendingAdjustment += priorCount - removed
Review Comment:
**🔴 Critical**: `accumulatingPendingAdjustment` (returned as
`pendingAdjustment` from `startEventTimeBundle` and
`startProcessingTimeBundle`) can be negative, or non-zero even when
`len(toProcess) == 0`.
However, in `ElementManager.Bundles` (lines 487-489 and 512-514),
`pendingAdjustment` is only applied when `ok && pendingAdjustment > 0`:
```go
if ok {
if pendingAdjustment > 0 {
em.addPending(pendingAdjustment)
}
```
Consider what happens when an early pane fires in **discarding** mode (or
when a trigger finishes before any `Accumulated` elements are retained) after
`addPendingOnTimeLocked` added `+1` to `em.pendingElements`, or if
`pendingOnTime` is taken with `priorCount == 0`:
- `priorCount - removed` (or `priorCount + pendingOnTimeDelta` in
`buildTriggeredBundle`) evaluates to `0 - 1 = -1`.
- Because `Bundles()` checks `pendingAdjustment > 0`, the `-1` adjustment is
dropped when the bundle starts, or if `ok == false`.
- When the bundle finishes, `PersistBundle` calls
`em.addPending(-len(completed.es))`.
Even in the common accumulating case with `priorCount == 1` and `removed ==
1` (`accumulatingPendingAdjustment == 0`), if a single bundle processes **two**
windows where one has `priorCount == 0, removed == 1` and or in
`buildProcessingTimeBundle` where `endOfWindowReached` is true and `priorCount
== 0, pendingOnTimeDelta == -1`, `pendingAdjustment` is `-1`. Dropping negative
`pendingAdjustment` leaves `em.pendingElements` permanently off by 1 (causing
the pipeline to hang at quiescence with `nothing in progress and no refreshes
with non zero pending elements`), or in other cases causes `em.pendingElements`
to hit zero early / panic with negative WaitGroup counter.
`Bundles()` should apply any non-zero `pendingAdjustment != 0` regardless of
sign (and even when `!ok` if `takePendingOnTimeLocked` removed an entry without
producing elements).
##########
sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:
##########
@@ -1732,68 +1810,80 @@ func (ss *stageState) savePanes(bundID string,
panesInBundle []bundlePane) {
}
}
-func (ss *stageState) buildTriggeredBundle(em *ElementManager, key string, win
typex.Window) ([]element, int) {
- var toProcess []element
+func (ss *stageState) buildTriggeredBundle(key string, win typex.Window,
endOfWindowReached bool, holdsInBundle map[mtime.Time]int) ([]element, int) {
dnt := ss.pendingByKeys[key]
if dnt == nil {
- // If we set an after-processing-time trigger, but some other
triggers fire or
- // the end of window is reached before the first trigger could
fire, then
- // the pending elements are processed in other bundles, leaving
a nil when
- // we try to build this triggered bundle.
- return toProcess, 0
+ return nil, 0
}
+ var newElems []element
var notYet []element
- // Look at all elements for this key, and only for this window.
for dnt.elements.Len() > 0 {
e := heap.Pop(&dnt.elements).(element)
if e.window != win {
notYet = append(notYet, e)
continue
}
- toProcess = append(toProcess, e)
- }
-
- // accumulationDiff adjusts the pending elements count to include
- // the accumulated elements, which would be the new elements, but
- // also all previous elements as well, which are duplicated at
- // at every pane.
- accumulationDiff := 0
- if ss.strat.Accumulating {
- // When accumulating, we need to retain all elements until the
last pane firing.
- for _, e := range toProcess {
- heap.Push(&dnt.elements, e)
- }
- accumulationDiff += len(toProcess)
+ newElems = append(newElems, e)
}
dnt.elements = append(dnt.elements, notYet...)
if dnt.elements.Len() == 0 {
delete(ss.pendingByKeys, key)
} else {
- // Ensure the heap invariants are maintained.
heap.Init(&dnt.elements)
}
+ if len(newElems) == 0 {
+ return nil, 0
+ }
+
+ state := ss.state[LinkID{}][win][key]
+ priorCount := len(state.Accumulated)
+ toProcess := make([]element, 0, priorCount+len(newElems))
+ toProcess = append(toProcess, state.Accumulated...)
+ toProcess = append(toProcess, newElems...)
+
+ triggerFinished := ss.isTriggerFinished(win, key)
+ pendingOnTimeDelta := 0
+ if ss.strat.Accumulating && !triggerFinished {
+ nextAcc := make([]element, len(toProcess))
+ copy(nextAcc, toProcess)
+ state.Accumulated = nextAcc
+ if !endOfWindowReached && !ss.strat.IsNeverTrigger() {
Review Comment:
**🟡 Medium**: If a composite trigger such as `TriggerAfterAll` or
`TriggerAfterEach` or `TriggerOrFinally` is used in **discarding** mode
(`ss.strat.Accumulating == false`), an early firing does not call
`addPendingOnTimeLocked`. However, if an early firing happens and no more
elements arrive before the end of the window, Beam's trigger model still
requires an empty `ON_TIME` pane at the watermark unless the trigger is
finished (to produce the `ON_TIME` pane for the window). Conversely, if
`addPendingOnTimeLocked` is only for accumulating mode here, note that when a
later early firing finishes the trigger (`triggerFinished == true`) after an
earlier firing added `pendingOnTime`, `takePendingOnTimeLocked` is called here:
if `priorCount == 0`, `priorCount + pendingOnTimeDelta` becomes `-1`, which in
`injectTriggeredBundlesIfReady` is added to `count` and returned to
`PersistBundle` (`em.addPending(count)`), which works for `AddPending`, but in
`buildProcessingTimeBundle` (line 24
20) it is returned as `pendingAdjustment = -1` to `Bundles()`, where `if
pendingAdjustment > 0` ignores it!
##########
sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:
##########
@@ -2069,126 +2159,138 @@ func (ss *stageState) timeSortedDataReadyLocked(ts,
watermark mtime.Time) bool {
// buildEventTimeBundle for aggregation stages, processes all elements that
are within the watermark for completed windows.
func (*aggregateStageKind) buildEventTimeBundle(ss *stageState, watermark
mtime.Time) (toProcess elementHeap, _ mtime.Time, _ set[string], _
map[mtime.Time]int, panesInBundle []bundlePane, schedulable bool,
pendingAdjustment int) {
minTs := mtime.MaxTimestamp
- // TODO: Allow configurable limit of keys per bundle, and elements per
key to improve parallelism.
- // TODO: when we do, we need to ensure that the stage remains
schedualable for bundle execution, for remaining pending elements and keys.
- // With the greedy approach, we don't need to since "new data" triggers
a refresh, and so should completing processing of a bundle.
newKeys := set[string]{}
-
holdsInBundle := map[mtime.Time]int{}
-
accumulatingPendingAdjustment := 0
+
+ candidateKeys := set[string]{}
+ for k := range ss.pendingByKeys {
+ candidateKeys.insert(k)
+ }
+ for k := range ss.pendingOnTime {
+ candidateKeys.insert(k)
+ }
+
keysPerBundle:
- for k, dnt := range ss.pendingByKeys {
+ for k := range candidateKeys {
if ss.inprogressKeys.present(k) {
continue
}
- newKeys.insert(k)
- // Track the min-timestamp for later watermark handling.
- if dnt.elements[0].timestamp < minTs {
- minTs = dnt.elements[0].timestamp
- }
- var toProcessForKey []element
- var toSkipForKey []element
- // Can we pre-compute this bit when adding to pendingByKeys?
- // startBundle is in run in a single scheduling goroutine, so
moving per-element code
- // to be computed by the bundle parallel goroutines will speed
things up a touch.
- for dnt.elements.Len() > 0 {
- // For aggregations, only include elements for this key
- // if we're after the end of window, or after the
window expiry deadline.
- // We will only ever trigger aggregations by watermark
at most twice, once the watermark passes the window ends for OnTime completion,
- // and once for when the window is closing.
- elm := heap.Pop(&dnt.elements).(element)
- shouldSkip := false
-
- if watermark <= elm.window.MaxTimestamp() {
- // The watermark hasn't passed the end of the
window yet, we do nothing.
- shouldSkip = true
- }
- // Watermark is past the end of this window. Have we
fired an OnTime pane yet?
- state :=
ss.state[LinkID{}][elm.window][string(elm.keyBytes)]
- // If this is not the ontime firing for this key.
-
- if !shouldSkip && state.Pane.Timing != typex.PaneEarly
&& watermark <= ss.strat.EarliestCompletion(elm.window) {
- // The watermark is still before the earliest
final completion for this window.
- // Do not add further data for this firing.
- // If this is the Never trigger, we also don't
fire OnTime until after the earliest completion.
- shouldSkip = true
- }
- if !shouldSkip && ss.strat.IsNeverTrigger() &&
watermark <= ss.strat.EarliestCompletion(elm.window) {
- // The NeverTrigger only has a single firing at
the end of window + allowed lateness.
- shouldSkip = true
+ var firingWindows []typex.Window
+ firingSet := set[typex.Window]{}
+ var newByWin map[typex.Window][]element
+
+ if dnt := ss.pendingByKeys[k]; dnt != nil {
+ var toSkipForKey []element
+ for dnt.elements.Len() > 0 {
+ // Aggregate by watermark at most twice per
window: once at window end
+ // for OnTime, and once at window closing after
allowed lateness.
+ elm := heap.Pop(&dnt.elements).(element)
+ shouldSkip := false
+
+ if watermark <= elm.window.MaxTimestamp() {
+ shouldSkip = true
+ }
+ state := ss.state[LinkID{}][elm.window][k]
+ onTimePassed := state.Pane.Timing !=
typex.PaneEarly || elm.window.MaxTimestamp() < ss.output
+ if !shouldSkip && onTimePassed && watermark <=
ss.strat.EarliestCompletion(elm.window) {
+ shouldSkip = true
+ }
+ if !shouldSkip && ss.strat.IsNeverTrigger() &&
watermark <= ss.strat.EarliestCompletion(elm.window) {
+ shouldSkip = true
+ }
+
+ if shouldSkip {
+ toSkipForKey = append(toSkipForKey, elm)
+ continue
+ }
+ if newByWin == nil {
+ newByWin = map[typex.Window][]element{}
+ }
+ newByWin[elm.window] =
append(newByWin[elm.window], elm)
+ if !firingSet.present(elm.window) {
+ firingSet.insert(elm.window)
+ firingWindows = append(firingWindows,
elm.window)
+ }
}
- if shouldSkip {
- toSkipForKey = append(toSkipForKey, elm)
- } else {
- toProcessForKey = append(toProcessForKey, elm)
+ for _, elm := range toSkipForKey {
+ heap.Push(&dnt.elements, elm)
+ }
+ if dnt.elements.Len() == 0 {
+ delete(ss.pendingByKeys, k)
}
}
- // Re-add skipped elements to the heap.
- // This ensures that `dnt.elements` reflects only the truly
skipped elements for future processing.
- for _, elm := range toSkipForKey {
- heap.Push(&dnt.elements, elm)
+ for win := range ss.pendingOnTime[k] {
+ if watermark <= win.MaxTimestamp() {
+ continue
+ }
+ if ss.strat.IsNeverTrigger() && watermark <=
ss.strat.EarliestCompletion(win) {
+ continue
+ }
+ if !firingSet.present(win) {
+ firingSet.insert(win)
+ firingWindows = append(firingWindows, win)
+ }
}
- // Get the pane for the aggregation correct, only mutate it
once per key and window.
- handledWindows := set[typex.Window]{}
- for _, elm := range toProcessForKey {
- state :=
ss.state[LinkID{}][elm.window][string(elm.keyBytes)]
- if handledWindows.present(elm.window) {
- // The pane is already correct for this key +
window + firing.
- if ss.strat.Accumulating && !state.Pane.IsLast {
- // If this isn't the last pane, then we
must add the element back to the pending store for subsequent firings.
- heap.Push(&dnt.elements, elm)
- accumulatingPendingAdjustment++
- }
+ var toProcessForKey []element
+ for _, win := range firingWindows {
+ ss.initAggStateLocked(win, k)
+ state := ss.state[LinkID{}][win][k]
+ removed := ss.takePendingOnTimeLocked(win, k,
holdsInBundle)
+ newElems := newByWin[win]
+ priorCount := len(state.Accumulated)
+ if priorCount+len(newElems) == 0 {
continue
}
- handledWindows.insert(elm.window)
- state.Pane = computeNextWatermarkPane(state.Pane)
- // Determine if this is the last pane.
- // Check if this is the post closing firing, which will
be the last one.
- // Unless it's the ontime pane, at which point it can
never be last.
- if watermark > ss.strat.EarliestCompletion(elm.window)
&& state.Pane.Timing != typex.PaneOnTime {
+ winElems := make([]element, 0, priorCount+len(newElems))
+ winElems = append(winElems, state.Accumulated...)
+ winElems = append(winElems, newElems...)
+
+ state.Pane = computeNextWatermarkPane(state.Pane,
win.MaxTimestamp() < ss.output)
Review Comment:
**🟠 High**: When `buildEventTimeBundle` consumes unemitted elements
(`newElems`) from `ss.pendingByKeys[k]` for an `ON_TIME` or closing watermark
firing, `state.Trigger` is never updated or reset for those consumed elements.
For example, in `early_count2_nonempty_ontime_then_late_count1` (where
`Early: ElementCount(2), Late: ElementCount(1)`):
1. Element `3` arrives before the watermark advances.
`injectTriggeredBundlesIfReady` runs `Early.onElement`, incrementing `Early`'s
count to `1` (not ready) and setting `TriggerAfterEndOfWindow`'s `extra` to
`false` (`endOfWindowReached == false`).
2. Watermark advances past the window end (`20`). `buildEventTimeBundle`
pops element `3` and emits the `ON_TIME` pane `[1, 2, 3]`, but never resets
`Early` or transitions `TriggerAfterEndOfWindow`.
3. What if the trigger was `Repeatedly(ElementCount(2))` or `AfterEach` or
`OrFinally` instead of `AfterEndOfWindow`? Because `Early` / the subtrigger
still has `count = 1` from element `3` (which was already emitted in the
`ON_TIME` pane!), the very next late element `4` will see `count = 1 + 1 = 2`
and fire immediately on a single late element even though `ElementCount(2)`
required 2 new elements after the `ON_TIME` firing.
##########
sdks/go/pkg/beam/runners/prism/internal/handlerunner_test.go:
##########
@@ -16,13 +16,161 @@
package internal
import (
+ "context"
+ "fmt"
+ "os"
+ "slices"
+ "strings"
"testing"
+ "time"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window/trigger"
pipepb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/pipeline_v1"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/options/jobopts"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+
"github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/jobservices"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/runners/universal"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/testing/passert"
+ "github.com/apache/beam/sdks/v2/go/pkg/beam/testing/teststream"
"github.com/google/go-cmp/cmp"
"google.golang.org/protobuf/testing/protocmp"
)
+func init() {
+ register.Function4x0(formatAccumPaneFn)
+ register.Iter1[int64]()
+ register.Emitter1[string]()
+}
+
+func formatAccumPaneFn(pn beam.PaneInfo, _ int, iter func(*int64) bool, emit
func(string)) {
+ var vals []int64
+ var v int64
+ for iter(&v) {
+ vals = append(vals, v)
+ }
+ slices.Sort(vals)
+ emit(fmt.Sprintf("timing=%s index=%d nonSpec=%d first=%t last=%t
vals=%v",
+ pn.Timing, pn.Index, pn.NonSpeculativeIndex, pn.IsFirst,
pn.IsLast, vals))
+}
+
+func runLocalTestPipeline(t *testing.T, p *beam.Pipeline, runFn
func(*jobservices.Job)) {
+ t.Helper()
+ prevEndpoint := *jobopts.Endpoint
+ prevEnv := *jobopts.EnvironmentType
+ prevBin := *jobopts.WorkerBinary
+ prevJob := *jobopts.JobName
+
+ s := jobservices.NewServer(0, runFn)
+ go s.Serve()
+
+ f, err := os.CreateTemp("", "prism-worker-dummy")
+ if err != nil {
+ s.Stop()
+ t.Fatal(err)
+ }
Review Comment:
**🟢 Low**: The temporary file `f` created by `os.CreateTemp` is never closed
before being removed in `t.Cleanup`, leaking a file descriptor for the duration
of the test (and preventing removal on Windows if ever run there). Close `f`
right after creation.
```suggestion
_ = f.Close()
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
