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 a5b5a310c9429e6f50ed04c135217fcadea86bee Author: João Ferreira <[email protected]> AuthorDate: Fri Aug 18 19:55:46 2023 +0100 use separate thread --- .../kinesis/impl/KinesisSchedulerSourceStage.scala | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/impl/KinesisSchedulerSourceStage.scala b/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/impl/KinesisSchedulerSourceStage.scala index 0518f3d0c..5c520c8ae 100644 --- a/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/impl/KinesisSchedulerSourceStage.scala +++ b/kinesis/src/main/scala/org/apache/pekko/stream/connectors/kinesis/impl/KinesisSchedulerSourceStage.scala @@ -26,7 +26,7 @@ import software.amazon.kinesis.processor.{ ShardRecordProcessor, ShardRecordProc import scala.annotation.tailrec import scala.collection.mutable -import scala.concurrent.{ ExecutionContext, Future, Promise } +import scala.concurrent.{ Future, Promise } import scala.util.{ Failure, Success, Try } /** @@ -76,15 +76,19 @@ private[kinesis] class KinesisSchedulerSourceStage( private[this] val buffer = mutable.Queue.empty[CommittableRecord] private[this] var schedulerOpt: Option[Scheduler] = None - implicit def ec: ExecutionContext = materializer.executionContext - override def preStart(): Unit = { val scheduler = schedulerBuilder(new ShardRecordProcessorFactory { override def shardRecordProcessor(): ShardRecordProcessor = new ShardProcessor(newRecordCallback) }) + //Run the scheduler loop in a separate thread + val thread = new Thread(() => { + val result = Try {scheduler.run()} + callback.invoke(SchedulerShutdown(result)) + }, s"KinesisSchedulerSource") + thread.setDaemon(true) + thread.start() schedulerOpt = Some(scheduler) - Future(scheduler.run()).onComplete(result => callback.invoke(SchedulerShutdown(result))) matValue.success(scheduler) } private val callback: AsyncCallback[Command] = getAsyncCallback(awaitingRecords) @@ -113,6 +117,6 @@ private[kinesis] class KinesisSchedulerSourceStage( failStage(SchedulerUnexpectedShutdown(e)) } override def postStop(): Unit = - schedulerOpt.foreach(scheduler => Future(if (!scheduler.shutdownComplete()) scheduler.shutdown())) + schedulerOpt.foreach(scheduler => Future(if (!scheduler.shutdownComplete()) scheduler.shutdown())(materializer.executionContext)) } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
