mynameborat commented on a change in pull request #912: SEP-19 : Refactoring
sideInputs from SamzaContainer to ContainerStorageManager
URL: https://github.com/apache/samza/pull/912#discussion_r259967789
##########
File path:
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java
##########
@@ -364,14 +615,137 @@ public void start() throws SamzaException {
executorService.shutdown();
- // Stop consumers
- this.systemConsumers.values().forEach(systemConsumer ->
systemConsumer.stop());
+ // Stop store consumers
+ this.storeConsumers.values().forEach(systemConsumer ->
systemConsumer.stop());
// Now re-create persistent stores in read-write mode, leave
non-persistent stores as-is
recreatePersistentTaskStoresInReadWriteMode(this.containerModel,
jobContext, containerContext,
- storageEngineFactories, changelogSystemStreams, serdes,
taskInstanceMetrics, taskInstanceCollectors);
+ storageEngineFactories, serdes, taskInstanceMetrics,
taskInstanceCollectors);
- LOG.info("Restore complete");
+ LOG.info("Store Restore complete");
+ }
+
+ // Read sideInputs until all sideInputStreams are caughtup, so start() can
return
+ private void startSideInputs() {
+
+ LOG.info("SideInput Restore started");
+
+ // initialize the sideInputStorageManagers
+ getSideInputStorageManagers().forEach(sideInputStorageManager ->
sideInputStorageManager.init());
+
+ // start the checkpointing thread at the commit-ms frequency
+ sideInputsFlushFuture = sideInputsFlushExecutor.scheduleWithFixedDelay(new
Runnable() {
+ @Override
+ public void run() {
+ getSideInputStorageManagers().forEach(sideInputStorageManager ->
sideInputStorageManager.flush());
Review comment:
what happens when flush throws an exception? Can we also capture that into
sideInputException?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
