shunping commented on code in PR #38736:
URL: https://github.com/apache/beam/pull/38736#discussion_r3325897889
##########
sdks/go/pkg/beam/core/runtime/exec/datasampler_test.go:
##########
@@ -104,16 +104,9 @@ func TestDataSampler(t *testing.T) {
for _, sample := range test.samples {
dataSampler.SendSample(sample.PCollectionID,
sample.Element, sample.Timestamp)
}
- var samplesCount = -1
- var samples map[string][]*DataSample
- for i := 0; i < 5; i++ {
- samples = dataSampler.GetSamples(test.pids)
- if len(samples) == len(test.want) {
- samplesCount = len(samples)
- break
- }
- time.Sleep(time.Second)
- }
+ time.Sleep(1 * time.Second)
+ samples := dataSampler.GetSamples(test.pids)
+ samplesCount := len(samples)
Review Comment:
Ack.
The 1-second sleep acts as a hard timeout to ensure we don't miss anything.
Gemini's proposal is only a more efficient implementation of that same timeout,
where it allows to finish faster if all samples arrive early.
The core issue is that we don't have a deterministic signal for when all
asynchronous samples have finished processing. And it remains unsolved here.
--
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]
