Make access to the java consumer fair

[flavienbert]

Replace the semaphore that is used to control access to the java consumer with a reentrant lock that has fairness enabled. This prevents liveness problems for consumers that need to use the consumer, e.g. to fetch committed offsets. Fixes #1107.

[flavienbert]

Fixes #1107

[svroonland]

Have you verified that this fixes the issue? A unit test would of course be very welcome

[flavienbert]

I confirm it fixes the issue. And it fixes an other issue: #1036 I don't have the RejectedExecutionException anymore and the process close normally! Do you have an idea of this fix? Probably the java consumer is not used in time during shutdow process?

 ZIO
      .blocking(ZIO.suspend(f(consumer)))
      .catchSome { case _: WakeupException =>
        ZIO.interrupt
      }
      .fork
      .flatMap(fib => fib.join.onInterrupt(ZIO.succeed(consumer.wakeup()) *> fib.interrupt))

[erikvanoosten]

I like it!

So now the question is, how to unit test this? Perhaps by calling to the consumer from the stream (e.g. get offsets), and then verifying that it makes progress?

[erikvanoosten]

Probably the java consumer is not used in time during shutdow process?

Likely.

[flavienbert]

I tried to run the test I wrote with the semaphore version, and the test fail.

[erikvanoosten]

Thanks @flavienbert !

[flavienbert]

@erikvanoosten you're welcome. Do you plan to publish a new version with this fix soon?

[erikvanoosten]

Yes definitely! Probably the coming weekend.

[erikvanoosten]

@flavienbert @erikvanoosten Seeing this PR from ZIO (merged but not yet released): zio/zio#8554 and seeing that zio-concurrent repo is archived, do we really want/need to use zio-concurrent ReentrantLock?

Edit: I checked and ReentrantLock is directly present in ZIO. We don't need the zio-concurrent dependency

Zio-concurrent is now a module in the main zio repository.
Eh, I do not see it in ZIO itself... did I miss something?

[guizmaii]

@erikvanoosten My bad. I though it was coming from the zio-concurrent repo but comes indeed from ZIO's main repo

[erikvanoosten]

Now that the standard semaphore in zio is fair (since 2.0.20, see zio/zio#8554), we can revert this change, allowing us to remove an external dependency.