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 681d9e6db8b876667000b41c36e002c0a5fef2af Author: João Ferreira <[email protected]> AuthorDate: Wed Dec 27 15:19:02 2023 +0000 Revert "using a separate thread works" This reverts commit 1aeeac1ca42667a0ca704370a7c0fe0e9ad9d0e6. --- .../kinesis/impl/KinesisSchedulerSourceStage.scala | 26 +++++++++++++++------- 1 file changed, 18 insertions(+), 8 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 b72922b1f..fa1b6b3fe 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.{ Future, Promise } +import scala.concurrent.{ ExecutionContext, Future, Promise } import scala.util.{ Failure, Success, Try } /** @@ -77,18 +77,13 @@ private[kinesis] class KinesisSchedulerSourceStage( private[this] var schedulerOpt: Option[Scheduler] = None override def preStart(): Unit = { + implicit val ec: ExecutionContext = executionContext(attributes) 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) @@ -119,5 +114,20 @@ private[kinesis] class KinesisSchedulerSourceStage( override def postStop(): Unit = schedulerOpt.foreach(scheduler => Future(if (!scheduler.shutdownComplete()) scheduler.shutdown())(materializer.executionContext)) + + protected def executionContext(attributes: Attributes): ExecutionContext = { + val dispatcherId = (attributes.get[ActorAttributes.Dispatcher](ActorAttributes.IODispatcher) match { + case ActorAttributes.Dispatcher("") => + ActorAttributes.IODispatcher + case d => d + }) match { + case d @ ActorAttributes.IODispatcher => + // this one is not a dispatcher id, but is a config path pointing to the dispatcher id + materializer.system.settings.config.getString(d.dispatcher) + case d => d.dispatcher + } + + materializer.system.dispatchers.lookup(dispatcherId) + } } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
