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_r259966131
 
 

 ##########
 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());
+      }
+    }, 0, new TaskConfig(config).getCommitMs(), TimeUnit.MILLISECONDS);
+
+    // set the latch to the number of sideInput SSPs
+    this.sideInputsCaughtUp = new 
CountDownLatch(this.sideInputStorageManagers.keySet().size());
+
+    // register all side input SSPs with the consumers
+    for (SystemStreamPartition ssp : sideInputStorageManagers.keySet()) {
+      String startingOffset = 
sideInputStorageManagers.get(ssp).getStartingOffset(ssp);
+
+      if (startingOffset == null) {
+        throw new SamzaException("No offset defined for SideInput 
SystemStreamPartition : " + ssp);
+      }
+
+      // register startingOffset with the sysConsumer and register a metric 
for it
+      sideInputSystemConsumers.register(ssp, startingOffset, null);
+      
taskInstanceMetrics.get(sideInputStorageManagers.get(ssp).getTaskName()).addOffsetGauge(
+          ssp, ScalaJavaUtil.toScalaFunction(() -> 
sideInputStorageManagers.get(ssp).getLastProcessedOffset(ssp)));
+
+      SystemStreamMetadata systemStreamMetadata = 
streamMetadataCache.getSystemStreamMetadata(ssp.getSystemStream(), false);
+      SystemStreamMetadata.SystemStreamPartitionMetadata sspMetadata =
+          (systemStreamMetadata == null) ? null : 
systemStreamMetadata.getSystemStreamPartitionMetadata().get(ssp.getPartition());
+
+      // record a copy of the sspMetadata, to later check if its caught up
+      initialSideInputSSPMetadata.put(ssp, sspMetadata);
+
+      // check if the ssp is caught to upcoming, even at start
+      checkSideInputCaughtUp(ssp, startingOffset, 
SystemStreamMetadata.OffsetType.UPCOMING, false);
+    }
+
+    // start the systemConsumers for consuming input
+    this.sideInputSystemConsumers.start();
+
+    // create a thread for sideInput reads
+    Thread readSideInputs = new Thread(() -> {
+        while (!shutDownSideInputRead) {
+          IncomingMessageEnvelope envelope = 
sideInputSystemConsumers.choose(true);
+          if (envelope != null) {
+
+            if (!envelope.isEndOfStream())
+              
sideInputStorageManagers.get(envelope.getSystemStreamPartition()).process(envelope);
+
+            checkSideInputCaughtUp(envelope.getSystemStreamPartition(), 
envelope.getOffset(),
+                SystemStreamMetadata.OffsetType.NEWEST, 
envelope.isEndOfStream());
+
+          } else {
+            LOG.trace("No incoming message was available");
+          }
+        }
+      });
+
+    readSideInputs.setUncaughtExceptionHandler(new 
Thread.UncaughtExceptionHandler() {
+      @Override
+      public void uncaughtException(Thread t, Throwable e) {
+        sideInputException = Optional.of(e);
+        sideInputsCaughtUp.countDown();
 
 Review comment:
   I am not sure this will work in all cases. Attaching an uncaught exception 
handler to the thread will only make sure we capture the exception and 
countDown the latch once in the event of an exception. Since you are not using 
an executor service, the thread is not going to be created again and your main 
loop will still be stuck on the latch for the other SSPs
   
   

----------------------------------------------------------------
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

Reply via email to