@Test public final void testMultiStreamInitialization() throws ProvisionedThroughputException, DependencyException { retrievalConfig = new RetrievalConfig(kinesisClient, multiStreamTracker, applicationName) .retrievalFactory(retrievalFactory); leaseManagementConfig = new LeaseManagementConfig(tableName, dynamoDBClient, kinesisClient, workerIdentifier).leaseManagementFactory(new TestKinesisLeaseManagementFactory(true, true)); scheduler = new Scheduler(checkpointConfig, coordinatorConfig, leaseManagementConfig, lifecycleConfig, metricsConfig, processorConfig, retrievalConfig); scheduler.initialize(); shardDetectorMap.values().stream() .forEach(shardDetector -> verify(shardDetector, times(1)).listShards()); shardSyncTaskManagerMap.values().stream() .forEach(shardSyncTM -> verify(shardSyncTM, times(1)).hierarchicalShardSyncer()); }
@VisibleForTesting void initialize() { synchronized (lock) { registerErrorHandlerForUndeliverableAsyncTaskExceptions(); workerStateChangeListener.onWorkerStateChange(WorkerStateChangeListener.WorkerState.INITIALIZING); boolean isDone = false; Exception lastException = null; for (int i = 0; (!isDone) && (i < maxInitializationAttempts); i++) { try { log.info("Initialization attempt {}", (i + 1)); log.info("Initializing LeaseCoordinator"); leaseCoordinator.initialize(); TaskResult result; if (!skipShardSyncAtWorkerInitializationIfLeasesExist || leaseRefresher.isLeaseTableEmpty()) { if (shouldInitiateLeaseSync()) { log.info("Worker {} is initiating the lease sync.", leaseManagementConfig.workerIdentifier()); leaderElectedPeriodicShardSyncManager.syncShardsOnce(); } } else { log.info("Skipping shard sync per configuration setting (and lease table is not empty)"); } leaseCleanupManager.start(); if (!leaseCoordinator.isRunning()) { log.info("Starting LeaseCoordinator"); leaseCoordinator.start(); } else { log.info("LeaseCoordinator is already running. No need to start it."); } log.info("Scheduling periodicShardSync"); leaderElectedPeriodicShardSyncManager.start(); streamSyncWatch.start(); isDone = true; } catch (LeasingException e) { log.error("Caught exception when initializing LeaseCoordinator", e); lastException = e; } catch (Exception e) { lastException = e; } if (!isDone) { try { Thread.sleep(schedulerInitializationBackoffTimeMillis); leaderElectedPeriodicShardSyncManager.stop(); } catch (InterruptedException e) { log.debug("Sleep interrupted while initializing worker."); } } } if (!isDone) { throw new RuntimeException(lastException); } workerStateChangeListener.onWorkerStateChange(WorkerStateChangeListener.WorkerState.STARTED); } }
Scheduler implements Runnable { @VisibleForTesting void initialize() { synchronized (lock) { registerErrorHandlerForUndeliverableAsyncTaskExceptions(); workerStateChangeListener.onWorkerStateChange(WorkerStateChangeListener.WorkerState.INITIALIZING); boolean isDone = false; Exception lastException = null; for (int i = 0; (!isDone) && (i < maxInitializationAttempts); i++) { try { log.info("Initialization attempt {}", (i + 1)); log.info("Initializing LeaseCoordinator"); leaseCoordinator.initialize(); TaskResult result; if (!skipShardSyncAtWorkerInitializationIfLeasesExist || leaseRefresher.isLeaseTableEmpty()) { if (shouldInitiateLeaseSync()) { log.info("Worker {} is initiating the lease sync.", leaseManagementConfig.workerIdentifier()); leaderElectedPeriodicShardSyncManager.syncShardsOnce(); } } else { log.info("Skipping shard sync per configuration setting (and lease table is not empty)"); } leaseCleanupManager.start(); if (!leaseCoordinator.isRunning()) { log.info("Starting LeaseCoordinator"); leaseCoordinator.start(); } else { log.info("LeaseCoordinator is already running. No need to start it."); } log.info("Scheduling periodicShardSync"); leaderElectedPeriodicShardSyncManager.start(); streamSyncWatch.start(); isDone = true; } catch (LeasingException e) { log.error("Caught exception when initializing LeaseCoordinator", e); lastException = e; } catch (Exception e) { lastException = e; } if (!isDone) { try { Thread.sleep(schedulerInitializationBackoffTimeMillis); leaderElectedPeriodicShardSyncManager.stop(); } catch (InterruptedException e) { log.debug("Sleep interrupted while initializing worker."); } } } if (!isDone) { throw new RuntimeException(lastException); } workerStateChangeListener.onWorkerStateChange(WorkerStateChangeListener.WorkerState.STARTED); } } }
Scheduler implements Runnable { @VisibleForTesting void initialize() { synchronized (lock) { registerErrorHandlerForUndeliverableAsyncTaskExceptions(); workerStateChangeListener.onWorkerStateChange(WorkerStateChangeListener.WorkerState.INITIALIZING); boolean isDone = false; Exception lastException = null; for (int i = 0; (!isDone) && (i < maxInitializationAttempts); i++) { try { log.info("Initialization attempt {}", (i + 1)); log.info("Initializing LeaseCoordinator"); leaseCoordinator.initialize(); TaskResult result; if (!skipShardSyncAtWorkerInitializationIfLeasesExist || leaseRefresher.isLeaseTableEmpty()) { if (shouldInitiateLeaseSync()) { log.info("Worker {} is initiating the lease sync.", leaseManagementConfig.workerIdentifier()); leaderElectedPeriodicShardSyncManager.syncShardsOnce(); } } else { log.info("Skipping shard sync per configuration setting (and lease table is not empty)"); } leaseCleanupManager.start(); if (!leaseCoordinator.isRunning()) { log.info("Starting LeaseCoordinator"); leaseCoordinator.start(); } else { log.info("LeaseCoordinator is already running. No need to start it."); } log.info("Scheduling periodicShardSync"); leaderElectedPeriodicShardSyncManager.start(); streamSyncWatch.start(); isDone = true; } catch (LeasingException e) { log.error("Caught exception when initializing LeaseCoordinator", e); lastException = e; } catch (Exception e) { lastException = e; } if (!isDone) { try { Thread.sleep(schedulerInitializationBackoffTimeMillis); leaderElectedPeriodicShardSyncManager.stop(); } catch (InterruptedException e) { log.debug("Sleep interrupted while initializing worker."); } } } if (!isDone) { throw new RuntimeException(lastException); } workerStateChangeListener.onWorkerStateChange(WorkerStateChangeListener.WorkerState.STARTED); } } Scheduler(@NonNull final CheckpointConfig checkpointConfig,
                     @NonNull final CoordinatorConfig coordinatorConfig,
                     @NonNull final LeaseManagementConfig leaseManagementConfig,
                     @NonNull final LifecycleConfig lifecycleConfig,
                     @NonNull final MetricsConfig metricsConfig,
                     @NonNull final ProcessorConfig processorConfig,
                     @NonNull final RetrievalConfig retrievalConfig); @VisibleForTesting protected  Scheduler(@NonNull final CheckpointConfig checkpointConfig,
                        @NonNull final CoordinatorConfig coordinatorConfig,
                        @NonNull final LeaseManagementConfig leaseManagementConfig,
                        @NonNull final LifecycleConfig lifecycleConfig,
                        @NonNull final MetricsConfig metricsConfig,
                        @NonNull final ProcessorConfig processorConfig,
                        @NonNull final RetrievalConfig retrievalConfig,
                        @NonNull final DiagnosticEventFactory diagnosticEventFactory); }
Scheduler implements Runnable { @VisibleForTesting void initialize() { synchronized (lock) { registerErrorHandlerForUndeliverableAsyncTaskExceptions(); workerStateChangeListener.onWorkerStateChange(WorkerStateChangeListener.WorkerState.INITIALIZING); boolean isDone = false; Exception lastException = null; for (int i = 0; (!isDone) && (i < maxInitializationAttempts); i++) { try { log.info("Initialization attempt {}", (i + 1)); log.info("Initializing LeaseCoordinator"); leaseCoordinator.initialize(); TaskResult result; if (!skipShardSyncAtWorkerInitializationIfLeasesExist || leaseRefresher.isLeaseTableEmpty()) { if (shouldInitiateLeaseSync()) { log.info("Worker {} is initiating the lease sync.", leaseManagementConfig.workerIdentifier()); leaderElectedPeriodicShardSyncManager.syncShardsOnce(); } } else { log.info("Skipping shard sync per configuration setting (and lease table is not empty)"); } leaseCleanupManager.start(); if (!leaseCoordinator.isRunning()) { log.info("Starting LeaseCoordinator"); leaseCoordinator.start(); } else { log.info("LeaseCoordinator is already running. No need to start it."); } log.info("Scheduling periodicShardSync"); leaderElectedPeriodicShardSyncManager.start(); streamSyncWatch.start(); isDone = true; } catch (LeasingException e) { log.error("Caught exception when initializing LeaseCoordinator", e); lastException = e; } catch (Exception e) { lastException = e; } if (!isDone) { try { Thread.sleep(schedulerInitializationBackoffTimeMillis); leaderElectedPeriodicShardSyncManager.stop(); } catch (InterruptedException e) { log.debug("Sleep interrupted while initializing worker."); } } } if (!isDone) { throw new RuntimeException(lastException); } workerStateChangeListener.onWorkerStateChange(WorkerStateChangeListener.WorkerState.STARTED); } } Scheduler(@NonNull final CheckpointConfig checkpointConfig,
                     @NonNull final CoordinatorConfig coordinatorConfig,
                     @NonNull final LeaseManagementConfig leaseManagementConfig,
                     @NonNull final LifecycleConfig lifecycleConfig,
                     @NonNull final MetricsConfig metricsConfig,
                     @NonNull final ProcessorConfig processorConfig,
                     @NonNull final RetrievalConfig retrievalConfig); @VisibleForTesting protected  Scheduler(@NonNull final CheckpointConfig checkpointConfig,
                        @NonNull final CoordinatorConfig coordinatorConfig,
                        @NonNull final LeaseManagementConfig leaseManagementConfig,
                        @NonNull final LifecycleConfig lifecycleConfig,
                        @NonNull final MetricsConfig metricsConfig,
                        @NonNull final ProcessorConfig processorConfig,
                        @NonNull final RetrievalConfig retrievalConfig,
                        @NonNull final DiagnosticEventFactory diagnosticEventFactory); @Override void run(); Future<Boolean> startGracefulShutdown(); Callable<Boolean> createGracefulShutdownCallable(); boolean hasGracefulShutdownStarted(); void shutdown(); @Deprecated Future<Void> requestShutdown(); }
Scheduler implements Runnable { @VisibleForTesting void initialize() { synchronized (lock) { registerErrorHandlerForUndeliverableAsyncTaskExceptions(); workerStateChangeListener.onWorkerStateChange(WorkerStateChangeListener.WorkerState.INITIALIZING); boolean isDone = false; Exception lastException = null; for (int i = 0; (!isDone) && (i < maxInitializationAttempts); i++) { try { log.info("Initialization attempt {}", (i + 1)); log.info("Initializing LeaseCoordinator"); leaseCoordinator.initialize(); TaskResult result; if (!skipShardSyncAtWorkerInitializationIfLeasesExist || leaseRefresher.isLeaseTableEmpty()) { if (shouldInitiateLeaseSync()) { log.info("Worker {} is initiating the lease sync.", leaseManagementConfig.workerIdentifier()); leaderElectedPeriodicShardSyncManager.syncShardsOnce(); } } else { log.info("Skipping shard sync per configuration setting (and lease table is not empty)"); } leaseCleanupManager.start(); if (!leaseCoordinator.isRunning()) { log.info("Starting LeaseCoordinator"); leaseCoordinator.start(); } else { log.info("LeaseCoordinator is already running. No need to start it."); } log.info("Scheduling periodicShardSync"); leaderElectedPeriodicShardSyncManager.start(); streamSyncWatch.start(); isDone = true; } catch (LeasingException e) { log.error("Caught exception when initializing LeaseCoordinator", e); lastException = e; } catch (Exception e) { lastException = e; } if (!isDone) { try { Thread.sleep(schedulerInitializationBackoffTimeMillis); leaderElectedPeriodicShardSyncManager.stop(); } catch (InterruptedException e) { log.debug("Sleep interrupted while initializing worker."); } } } if (!isDone) { throw new RuntimeException(lastException); } workerStateChangeListener.onWorkerStateChange(WorkerStateChangeListener.WorkerState.STARTED); } } Scheduler(@NonNull final CheckpointConfig checkpointConfig,
                     @NonNull final CoordinatorConfig coordinatorConfig,
                     @NonNull final LeaseManagementConfig leaseManagementConfig,
                     @NonNull final LifecycleConfig lifecycleConfig,
                     @NonNull final MetricsConfig metricsConfig,
                     @NonNull final ProcessorConfig processorConfig,
                     @NonNull final RetrievalConfig retrievalConfig); @VisibleForTesting protected  Scheduler(@NonNull final CheckpointConfig checkpointConfig,
                        @NonNull final CoordinatorConfig coordinatorConfig,
                        @NonNull final LeaseManagementConfig leaseManagementConfig,
                        @NonNull final LifecycleConfig lifecycleConfig,
                        @NonNull final MetricsConfig metricsConfig,
                        @NonNull final ProcessorConfig processorConfig,
                        @NonNull final RetrievalConfig retrievalConfig,
                        @NonNull final DiagnosticEventFactory diagnosticEventFactory); @Override void run(); Future<Boolean> startGracefulShutdown(); Callable<Boolean> createGracefulShutdownCallable(); boolean hasGracefulShutdownStarted(); void shutdown(); @Deprecated Future<Void> requestShutdown();  }