Processing happens in a separate goroutine so even if we cancel the context as soon as we start processing the first batch of records a second fetch might have completed already.
The other scenario is that the context is canceled before the second fetch starts creating a +- tc.maxPollRecords variance in errored records and potentially causing extra records in processed record counter. Update the test to ignore extra records processed after the context has been canceled and account for an extra tc.maxPollRecords records when asserting processed records.
Make the test more deterministic and assert for equality instead of comparing >=.
This issue only happens if tc.initialRecords != tc.maxPollRecords.
Mostly affecting 30_produced_2_poll_AMOD.
Example:
Scenario 1:
produce 30 records
poll 2 records
process 2 records (async)
poll 2 records
context canceled (async when processing the first 2 records)
process 2 records (async)
consumer stop
Scenario 2:
produce 30 records
poll 2 records
process 2 records (async)
context canceled (async when processing the first 2 records)
consumer stop
Scenario 3:
produce 30 records
poll 2 records
process 2 records (async)
poll 2 records
context canceled (async when processing the first 2 records)
consumer stop
To reproduce:
go test -run=TestConsumerDelivery ./kafka/ -count 100 -race -failfast
System-tests will not be executed for this PR because it is from a forked repository.
But it will still create a successful run-system-tests status check.
Processing happens in a separate goroutine so even if we cancel the context as soon as we start processing the first batch of records a second fetch might have completed already. The other scenario is that the context is canceled before the second fetch starts creating a +-
tc.maxPollRecords
variance in errored records and potentially causing extra records in processed record counter. Update the test to ignore extra records processed after the context has been canceled and account for an extra tc.maxPollRecords records when asserting processed records. Make the test more deterministic and assert for equality instead of comparing >=.This issue only happens if
tc.initialRecords
!=tc.maxPollRecords
.Mostly affecting
30_produced_2_poll_AMOD
.Example:
Scenario 1:
Scenario 2:
Scenario 3:
To reproduce:
go test -run=TestConsumerDelivery ./kafka/ -count 100 -race -failfast