Fix timing issue in ConsumerSpec test by adding delay in consumer stream

[AdrielC]

This PR addresses a timing issue in the ConsumerSpec test:

“it’s possible to start a new consumption session from a Consumer that had a consumption session stopped previously”

Issue:

When running the entire test suite, this test occasionally fails with the following assertion error:

Assertion failed:
  ✗ 100000 was not less than 100000
  consumed0 did not satisfy isGreaterThan(0L) && isLessThan(numberOfMessages.toLong)
  consumed0 = 100000

Cause:

• The failure occurs because the first consumer sometimes consumes all messages before consumer.stopConsumption is called.
• This happens due to timing variations when the test suite is run in full, possibly because of system performance or resource contention.
• As a result, consumed0 equals numberOfMessages, causing the assertion consumed0 < numberOfMessages.toLong to fail.

Solution:

• Introduce a slight delay in the consumer stream to prevent it from consuming all messages too quickly.
• This ensures that consumer.stopConsumption is called before all messages are consumed.
• The test can now reliably check that the consumer can be stopped and restarted.

Testing:

• Ran the full test suite multiple times to confirm that the issue is resolved.
• Verified that consumed0 is greater than 0 and less than numberOfMessages, satisfying the original assertions.

[CLAassistant]

All committers have signed the CLA.

[erikvanoosten]

Thanks for your contribution @AdrielC !

A simple but effective solution. The test is already waiting that 200 ms anyway, so this will not make the test go any slower.

Will you remove lines 1483 - 1485 please?

[svroonland]

There's a number of zio-kafka tests that are doing sleeps to wait for some condition to occur. In my experience those are always flaky on CI at some point.

A better solution is to use a Promise to signal that some condition has occurred and await it somewhere else. By creating a Promise[Nothing, Unit], adding something like .tap(_ => promise.succeed(())) and adding a promise.await later on, we can be sure that the condition has occurred. Have a look at the can stop one stream while keeping another one running test for example.

If you'd like to implement that for this test, that would be great. If not for now, we can merge this PR and create an issue to improve on all usages of sleep in the tests.

[erikvanoosten]

If you'd like to implement that for this test, that would be great. If not for now, we can merge this PR and create an issue to improve on all usages of sleep in the tests.

We should do the second regardless of what @AdrielC want to do for now 🙂