This is an automated email from the ASF dual-hosted git repository. mdedetrich pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/incubator-pekko-connectors.git
commit 4eb10dda9b3dae31fc80aa380b2efbc16918603a Author: João Ferreira <[email protected]> AuthorDate: Mon Aug 21 23:48:11 2023 +0100 test showing the thread starvation --- .../kinesis/KinesisSchedulerSourceSpec.scala | 36 ++++++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/kinesis/src/test/scala/org/apache/pekko/stream/connectors/kinesis/KinesisSchedulerSourceSpec.scala b/kinesis/src/test/scala/org/apache/pekko/stream/connectors/kinesis/KinesisSchedulerSourceSpec.scala index bca357c36..5cfb66c63 100644 --- a/kinesis/src/test/scala/org/apache/pekko/stream/connectors/kinesis/KinesisSchedulerSourceSpec.scala +++ b/kinesis/src/test/scala/org/apache/pekko/stream/connectors/kinesis/KinesisSchedulerSourceSpec.scala @@ -251,6 +251,42 @@ class KinesisSchedulerSourceSpec killSwitch.shutdown() sinkProbe.expectComplete() }) + + "not starve dispatcher threads with 15 sources" in assertAllStagesStopped { + val scheduler: Scheduler = org.mockito.Mockito.mock(classOf[Scheduler]) + when(scheduler.run()).thenAnswer(new Answer[Unit] { + override def answer(invocation: InvocationOnMock): Unit = + Thread.sleep(10000) + }) + + val killSwitch = KillSwitches.shared("") + + for (_ <- 1 to 15) + KinesisSchedulerSource(_ => scheduler, KinesisSchedulerSourceSettings.defaults) + .via(killSwitch.flow) + .run() + + killSwitch.shutdown() + + } + + "not starve dispatcher threads with 16 sources" in assertAllStagesStopped { + val scheduler: Scheduler = org.mockito.Mockito.mock(classOf[Scheduler]) + when(scheduler.run()).thenAnswer(new Answer[Unit] { + override def answer(invocation: InvocationOnMock): Unit = + Thread.sleep(10000) + }) + + val killSwitch = KillSwitches.shared("") + + for (_ <- 1 to 16) + KinesisSchedulerSource(_ => scheduler, KinesisSchedulerSourceSettings.defaults) + .via(killSwitch.flow) + .run() + + killSwitch.shutdown() + + } } private abstract class KinesisSchedulerContext(schedulerFailure: Option[Throwable] = None, --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
