KafkaSourceProviderImplSpec: The KafkaSourceProviderImpl must successfully verify offsets from assigned partitions

This test appears to be flaky:

```
--> [org.apache.pekko.projection.kafka.internal.KafkaSourceProviderImplSpec: The KafkaSourceProviderImpl must successfully verify offsets from assigned partitions] Start of log messages of test that [Failed(java.lang.AssertionError: timeout (3 seconds) while expecting 5 messages (got 4))]
[2024-02-07 19:15:36,982] [INFO] [org.apache.pekko.event.slf4j.Slf4jLogger] [] [KafkaSourceProviderImplSpec-pekko.actor.default-dispatcher-5] - Slf4jLogger started MDC: {}
[2024-02-07 19:15:37,249] [DEBUG] [KafkaSourceProviderImplSpec] [] [pool-6-thread-24] - Starting ActorTestKit MDC: {}
[2024-02-07 19:15:37,415] [DEBUG] [KafkaSourceProviderImplSpec] [] [pool-6-thread-24] - Starting ActorTestKit MDC: {}
[2024-02-07 19:15:37,606] [INFO] [org.apache.pekko.actor.testkit.typed.scaladsl.LogCapturing] [] [pool-6-thread-24-ScalaTest-running-KafkaSourceProviderImplSpec] - Logging started for test [org.apache.pekko.projection.kafka.internal.KafkaSourceProviderImplSpec: The KafkaSourceProviderImpl must successfully verify offsets from assigned partitions] MDC: {}
[2024-02-07 19:15:38,102] [WARN] [org.apache.pekko.projection.testkit.internal.TestInternalProjectionState] [] [KafkaSourceProviderImplSpec-pekko.actor.default-dispatcher-5] - Source provider instructed projection to skip offset [MergeableOffset(Map(topic-1 -> 5))] with reason: The offset contains Kafka topic partitions that were revoked or lost in a previous rebalance MDC: {pekkoUid=57598404356998700, sourceThread=KafkaSourceProviderImplSpec-pekko.actor.default-dispatcher-3, sourceActorSystem=KafkaSourceProviderImplSpec, pekkoAddress=pekko://KafkaSourceProviderImplSpec, pekkoSource=TestInternalProjectionState(pekko://KafkaSourceProviderImplSpec), pekkoTimestamp=19:15:38.101UTC}
[2024-02-07 19:15:38,104] [WARN] [org.apache.pekko.projection.testkit.internal.TestInternalProjectionState] [] [KafkaSourceProviderImplSpec-pekko.actor.default-dispatcher-5] - Source provider instructed projection to skip offset [MergeableOffset(Map(topic-1 -> 6))] with reason: The offset contains Kafka topic partitions that were revoked or lost in a previous rebalance MDC: {pekkoUid=57598404356998700, sourceThread=KafkaSourceProviderImplSpec-pekko.actor.default-dispatcher-3, sourceActorSystem=KafkaSourceProviderImplSpec, pekkoAddress=pekko://KafkaSourceProviderImplSpec, pekkoSource=TestInternalProjectionState(pekko://KafkaSourceProviderImplSpec), pekkoTimestamp=19:15:38.103UTC}
[2024-02-07 19:15:38,127] [WARN] [org.apache.pekko.projection.testkit.internal.TestInternalProjectionState] [] [KafkaSourceProviderImplSpec-pekko.actor.default-dispatcher-5] - Source provider instructed projection to skip offset [MergeableOffset(Map(topic-1 -> 7))] with reason: The offset contains Kafka topic partitions that were revoked or lost in a previous rebalance MDC: {pekkoUid=57598404356998700, sourceThread=KafkaSourceProviderImplSpec-pekko.actor.default-dispatcher-3, sourceActorSystem=KafkaSourceProviderImplSpec, pekkoAddress=pekko://KafkaSourceProviderImplSpec, pekkoSource=TestInternalProjectionState(pekko://KafkaSourceProviderImplSpec), pekkoTimestamp=19:15:38.127UTC}
[2024-02-07 19:15:38,128] [WARN] [org.apache.pekko.projection.testkit.internal.TestInternalProjectionState] [] [KafkaSourceProviderImplSpec-pekko.actor.default-dispatcher-5] - Source provider instructed projection to skip offset [MergeableOffset(Map(topic-1 -> 8))] with reason: The offset contains Kafka topic partitions that were revoked or lost in a previous rebalance MDC: {pekkoUid=57598404356998700, sourceThread=KafkaSourceProviderImplSpec-pekko.actor.default-dispatcher-3, sourceActorSystem=KafkaSourceProviderImplSpec, pekkoAddress=pekko://KafkaSourceProviderImplSpec, pekkoSource=TestInternalProjectionState(pekko://KafkaSourceProviderImplSpec), pekkoTimestamp=19:15:38.128UTC}
[2024-02-07 19:15:38,129] [WARN] [org.apache.pekko.projection.testkit.internal.TestInternalProjectionState] [] [KafkaSourceProviderImplSpec-pekko.actor.default-dispatcher-5] - Source provider instructed projection to skip offset [MergeableOffset(Map(topic-1 -> 9))] with reason: The offset contains Kafka topic partitions that were revoked or lost in a previous rebalance MDC: {pekkoUid=57598404356998700, sourceThread=KafkaSourceProviderImplSpec-pekko.actor.default-dispatcher-3, sourceActorSystem=KafkaSourceProviderImplSpec, pekkoAddress=pekko://KafkaSourceProviderImplSpec, pekkoSource=TestInternalProjectionState(pekko://KafkaSourceProviderImplSpec), pekkoTimestamp=19:15:38.129UTC}
[2024-02-07 19:15:38,132] [WARN] [org.apache.pekko.projection.testkit.internal.TestInternalProjectionState] [] [KafkaSourceProviderImplSpec-pekko.actor.default-dispatcher-5] - Source provider instructed projection to skip offset [MergeableOffset(Map(topic-1 -> 10))] with reason: The offset contains Kafka topic partitions that were revoked or lost in a previous rebalance MDC: {pekkoUid=57598404356998700, sourceThread=KafkaSourceProviderImplSpec-pekko.actor.default-dispatcher-3, sourceActorSystem=KafkaSourceProviderImplSpec, pekkoAddress=pekko://KafkaSourceProviderImplSpec, pekkoSource=TestInternalProjectionState(pekko://KafkaSourceProviderImplSpec), pekkoTimestamp=19:15:38.132UTC}
[2024-02-07 19:15:41,131] [INFO] [org.apache.pekko.actor.testkit.typed.scaladsl.LogCapturing] [] [pool-6-thread-24-ScalaTest-running-KafkaSourceProviderImplSpec] - Logging finished for test [org.apache.pekko.projection.kafka.internal.KafkaSourceProviderImplSpec: The KafkaSourceProviderImpl must successfully verify offsets from assigned partitions] that [Failed(java.lang.AssertionError: timeout (3 seconds) while expecting 5 messages (got 4))] MDC: {}
<-- [org.apache.pekko.projection.kafka.internal.KafkaSourceProviderImplSpec: The KafkaSourceProviderImpl must successfully verify offsets from assigned partitions] End of log messages of test that [Failed(java.lang.AssertionError: timeout (3 seconds) while expecting 5 messages (got 4))]
[info] - must successfully verify offsets from assigned partitions *** FAILED *** (3 seconds, 564 milliseconds)
[info] java.lang.AssertionError: timeout (3 seconds) while expecting 5 messages (got 4)
[info] at org.apache.pekko.actor.testkit.typed.internal.TestProbeImpl.assertFail(TestProbeImpl.scala:410)
[info] at org.apache.pekko.actor.testkit.typed.internal.TestProbeImpl.$anonfun$receiveMessages_internal$1(TestProbeImpl.scala:273)
[info] at org.scalatest.wordspec.AnyWordSpecLike.run(AnyWordSpecLike.scala:1353)
[info] at org.scalatest.wordspec.AnyWordSpecLike.run$(AnyWordSpecLike.scala:1351)
[info] at org.apache.pekko.projection.kafka.internal.KafkaSourceProviderImplSpec.run(KafkaSourceProviderImplSpec.scala:56)
[info] at org.scalatest.tools.Framework.org$scalatest$tools$Framework$$runSuite(Framework.scala:321)
[info] at org.scalatest.tools.Framework$ScalaTestTask.execute(Framework.scala:517)
[info] at sbt.TestRunner.runTest$1(TestFramework.scala:153)
[info] at sbt.TestRunner.run(TestFramework.scala:168)
[info] at sbt.TestFramework$$anon$3$$anonfun$$lessinit$greater$1.$anonfun$apply$1(TestFramework.scala:336)
[info] at sbt.TestFramework$.sbt$TestFramework$$withContextLoader(TestFramework.scala:296)
[info] at sbt.TestFramework$$anon$3$$anonfun$$lessinit$greater$1.apply(TestFramework.scala:336)
[info] at sbt.TestFramework$$anon$3$$anonfun$$lessinit$greater$1.apply(TestFramework.scala:336)
[info] at sbt.TestFunction.apply(TestFramework.scala:348)
[info] at sbt.Tests$.$anonfun$toTask$1(Tests.scala:436)
[info] at sbt.std.Transform$$anon$3.$anonfun$apply$2(Transform.scala:47)
[info] at sbt.std.Transform$$anon$4.work(Transform.scala:69)
[info] at sbt.Execute.$anonfun$submit$2(Execute.scala:283)
[info] at sbt.internal.util.ErrorHandling$.wideConvert(ErrorHandling.scala:24)
[info] at sbt.Execute.work(Execute.scala:292)
[info] at sbt.Execute.$anonfun$submit$1(Execute.scala:283)
[info] at sbt.ConcurrentRestrictions$$anon$4.$anonfun$submitValid$1(ConcurrentRestrictions.scala:265)
[info] at sbt.CompletionService$$anon$2.call(CompletionService.scala:65)
[info] at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
[info] at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
[info] at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
[info] at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
[info] at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
[info] at java.base/java.lang.Thread.run(Thread.java:829)
[info] Run completed in 5 seconds, 708 milliseconds.
[info] Total number of tests run: 1
[info] Suites: completed 1, aborted 0
[info] Tests: succeeded 0, failed 1, canceled 0, ignored 0, pending 0
[info] *** 1 TEST FAILED ***
[error] Failed tests:
[error] org.apache.pekko.projection.kafka.internal.KafkaSourceProviderImplSpec
```

Start with KafkaSourceProviderImplSpec.scala, especially the run method at line 56, and rerun the test named “must successfully verify offsets from assigned partitions.” Review the logged skipped offsets and rebalance messages; done means the test consistently receives all 5 expected messages without timing out.

Written by the indexing model from the issue text.