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 e710d4725f2183ad67322d9c7a32217b1f0f8ac1 Author: João Ferreira <[email protected]> AuthorDate: Mon Aug 21 23:52:29 2023 +0100 using a separate thread works --- .../kinesis/impl/KinesisSchedulerSourceStage.scala | 26 +++++++--------------- 1 file changed, 8 insertions(+), 18 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 fa1b6b3fe..b72922b1f 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 } /** @@ -77,13 +77,18 @@ 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) @@ -114,20 +119,5 @@ 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]
