Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -160,8 +160,10 @@ public void shutdown() throws IOException {
_shuttedDown.set(true);
_initializationtExecutorService.shutdownNow();
_synchronizer.stopPeriodicFetching();
_pushManager.stop();
_pushMonitorExecutorService.shutdownNow();
if (_streamingEnabledConfig.get()) {
_pushManager.stop();
_pushMonitorExecutorService.shutdownNow();
}
_segmentSynchronizationTaskImp.close();
_log.info("Successful shutdown of segment fetchers");
_splitSynchronizationTask.close();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@ public void startWithStreamingFalseShouldStartPolling() throws InterruptedExcept
Mockito.verify(_pushManager, Mockito.times(0)).start();

syncManager.shutdown();
Mockito.verify(_pushManager, Mockito.times(0)).stop();
Mockito.verify(_synchronizer, Mockito.times(1)).stopPeriodicDataRecording();
}

Expand All @@ -86,7 +87,7 @@ public void startWithStreamingTrueShouldStartSyncAll() throws InterruptedExcepti
}

@Test
public void onStreamingAvailable() throws InterruptedException, IOException {
public void onStreamingAvailable() throws InterruptedException {
TelemetryStorage telemetryStorage = Mockito.mock(TelemetryStorage.class);
LinkedBlockingQueue<PushManager.Status> messages = new LinkedBlockingQueue<>();
SplitTasks splitTasks = SplitTasks.build(_splitSynchronizationTask, _segmentSynchronizationTaskImp,
Expand All @@ -107,7 +108,7 @@ public void onStreamingAvailable() throws InterruptedException, IOException {
}

@Test
public void onStreamingDisabled() throws InterruptedException, IOException {
public void onStreamingDisabled() throws InterruptedException {
TelemetryStorage telemetryStorage = new InMemoryTelemetryStorage();
LinkedBlockingQueue<PushManager.Status> messsages = new LinkedBlockingQueue<>();
SplitTasks splitTasks = SplitTasks.build(_splitSynchronizationTask, _segmentSynchronizationTaskImp,
Expand All @@ -127,7 +128,7 @@ public void onStreamingDisabled() throws InterruptedException, IOException {
}

@Test
public void onStreamingShutdown() throws InterruptedException, IOException {
public void onStreamingShutdown() throws InterruptedException {
TelemetryStorage telemetryStorage = new InMemoryTelemetryStorage();
LinkedBlockingQueue<PushManager.Status> messsages = new LinkedBlockingQueue<>();
SplitTasks splitTasks = SplitTasks.build(_splitSynchronizationTask, _segmentSynchronizationTaskImp,
Expand Down Expand Up @@ -229,4 +230,4 @@ public void syncAllRetryThenShouldStartPolling() throws InterruptedException, IO
syncManager.shutdown();
Mockito.verify(_synchronizer, Mockito.times(1)).stopPeriodicDataRecording();
}
}
}