SEP-19 : Refactoring sideInputs from SamzaContainer to ContainerStorageManager #912

rmatharu-zz · 2019-02-08T03:42:55Z

This PR makes the following changes:

Moves the instantiation, wiring, of TaskSideInputStorageManagers from SamzaContainer to ContainerStorageManager.
Creates a separate SystemConsumers and set of sysConsumers in ContainerStorageManager which are responsible to read side inputs, but maintains existing bootstrap semantics for sideInputs.
Creates a periodic flush thread in ContainerStorageManager to flush all sideInputs, and made the TaskSideInputManager serialize access to process and flush.
Note that, currently both the sideInput reads (from SystemConsumers) and flushes are separate single thread.

Tested with existing unit and integration tests, sample side-input and non-sideInput apps on local VPC-dev cluster

… CSM.

…input

mynameborat

First pass. Will look into CSM again

mynameborat · 2019-02-19T17:05:40Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

@@ -101,6 +118,21 @@
  private final JobContext jobContext;
  private final ContainerContext containerContext;

+  /* Sideinput related parameters */
+  private final Map<String, List<SystemStream>> sideInputSystemStreams; // Map of side input system-streams indexed by store name


What happens when there are duplicate system stream associated with a store? Can this be Map<String, Set<SystemStram>> instead? By that, you explicitly disallow duplicates and be consistent with the semantics of sideInputSSps which has a task -> store -> set of ssps

mynameborat · 2019-02-19T17:15:55Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

+  private SystemConsumers sideInputSystemConsumers;
+  private final Map<SystemStreamPartition, SystemStreamMetadata.SystemStreamPartitionMetadata> initialSideInputSSPMetadata = new ConcurrentHashMap<>();
+  // Recorded sspMetadata of the sideInputSSPs recorded at start, used to determine when sideInputs are caughtup and container init can proceed
+  private CountDownLatch sideInputsCaughtUp; // Used by the sideInput-read thread to signal to the main thread


make it volatile

mynameborat · 2019-02-19T17:26:25Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

  }

  /**
   *  Creates SystemConsumer objects for store restoration, creating one consumer per system.
   */
-  private static Map<String, SystemConsumer> createStoreConsumers(Map<String, SystemStream> changelogSystemStreams,
+  private static Map<String, SystemConsumer> createConsumers(Map<String, List<SystemStream>> systemStreams,


Why does this have to take a Map<String, List<SystemStream>> systemStreams? It seems sufficient to just take in Set<SystemStreams>.

You can also get rid of the up conversion in the calling site for creating store consumers.

mynameborat · 2019-02-19T17:30:14Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java


  /** Maps containing relevant per-task objects */
  private final Map<TaskName, Map<String, StorageEngine>> taskStores;
  private final Map<TaskName, TaskRestoreManager> taskRestoreManagers;
  private final Map<TaskName, TaskInstanceMetrics> taskInstanceMetrics;
  private final Map<TaskName, TaskInstanceCollector> taskInstanceCollectors;

-  private final Map<String, SystemConsumer> systemConsumers; // Mapping from storeSystemNames to SystemConsumers
+  private final Map<String, SystemConsumer> storeConsumers; // Mapping from store name to SystemConsumers


Looks like this is a map from store system name to consumers from the creation path. Can you update the comment?

No its indexed by store name.

mynameborat · 2019-02-19T17:32:26Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

+  private final Map<TaskName, Map<String, Set<SystemStreamPartition>>> sideInputSSPs;
+  private final Map<TaskName, Map<String, SideInputsProcessor>> sideInputStoresToProcessor;
+  private final Map<SystemStreamPartition, TaskSideInputStorageManager> sideInputStorageManagers; // Map of sideInput storageManagers indexed by ssp, for simpler lookup for process()
+  private final Map<String, SystemConsumer> sideInputConsumers; // Mapping from storeSystemNames to SystemConsumers


Can we roll this into storeConsumers above? Store restoration and side inputs restoration happens sequentially, so not sure what is the purpose behind separating out the consumers.

The overlap of systems and system-consumers was a bit tricky to reason about so i let them the consumers be isolated.

mynameborat · 2019-02-19T20:45:50Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

@@ -101,6 +118,21 @@
  private final JobContext jobContext;
  private final ContainerContext containerContext;

+  /* Sideinput related parameters */
+  private final Map<String, List<SystemStream>> sideInputSystemStreams; // Map of side input system-streams indexed by store name
+  private final Map<TaskName, Map<String, Set<SystemStreamPartition>>> sideInputSSPs;


Cosmetics: I noticed you wrap side inputs related parameters within a task context or some map.

Can we rename them to make it clear. e.g. sideInputSSPs, makes me think its a set but its a map from task to a map of store to set of ssps.

mynameborat · 2019-02-19T20:51:47Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

+          });
+      });
+
+    // Create a map of changeLogSSP to storeName across all tasks, assuming no stores have the same changelogSSP


nit: "assuming no two stores"

mynameborat · 2019-02-19T21:04:52Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

+      });
+
+    // changelogSystemStreams correspond only to active tasks (since those of standby-tasks moved to side inputs above)
+    this.changelogSystemStreams = changelogSSPToStore.entrySet().stream().collect(Collectors.toMap(x -> x.getValue(), x -> x.getKey().getSystemStream(), (x, y) -> x));


iiuc, you are grouping the SSPs by store and projecting the SSPs to system stream.
you can also use collectors.groupingBy to simplify.

mynameborat · 2019-02-19T21:12:18Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

@@ -112,15 +144,49 @@

  public ContainerStorageManager(ContainerModel containerModel, StreamMetadataCache streamMetadataCache,


Cosmetics: Can we break down constructor functions? Logically group some of the initialization and separate them out to make it more readable.

mynameborat · 2019-02-19T22:01:52Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

-            storageEngine.stop();
-          });
+
+      Map<String, StorageEngine> persistentStores = this.taskStores.entrySet().stream().filter(e -> {


Will this stop side input stores? It is possible for side input stores to be persisted to disk.

No this wont, note that this is inside TaskRestoreManager, which only handles changelogged stores.

I noticed you use taskStores from the outer scope from CSM.
Can we instead use the helper method you have from outer class to get non-side-input stores to be safe?

…samza into test-sideinputrefactor

rmatharu-zz · 2019-02-22T04:20:35Z

Addressed all comments.
Thumbsup => the comment has been applied.

minor

…samza into test-sideinputrefactor

mynameborat

Another pass on the changes

mynameborat · 2019-02-25T17:48:11Z

samza-core/src/main/java/org/apache/samza/storage/StorageManagerUtil.java

@@ -207,9 +208,11 @@ public static boolean storeExists(File storeDir) {
   * @param storeBaseDir the base directory to use
   * @param storeName the store name to use
   * @param taskName the task name which is referencing the store
+   * @param taskMode


nit: add documentation

mynameborat · 2019-02-25T17:48:49Z

samza-core/src/main/java/org/apache/samza/storage/StorageManagerUtil.java

   * @return the partition directory for the store
   */
-  public static File getStorePartitionDir(File storeBaseDir, String storeName, TaskName taskName) {
+  public static File getStorePartitionDir(File storeBaseDir, String storeName, TaskName taskName, TaskMode taskMode) {
+    // TODO: use task-Mode to decide the storePartitionDir -- standby's dir should be the same as active


Are we doing this in a separate PR?

yes, i'll do a subsequent once when PR-903 is in

mynameborat · 2019-02-25T17:56:58Z

samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala

+    val sideInputSystemStreams = sideInputStoresToSystemStreams.values.flatMap(sideInputs => sideInputs.toStream).toSet
+
+    info("Got side input store system streams: %s" format sideInputSystemStreams)
+
    val inputSystemStreams = inputSystemStreamPartitions


Ideally, input SSPs should no longer have side inputs since its not using run loop anymore.
You can either clean up the side inputs wiring in high level in this PR or as a follow up PR. Also, we need to update the configuration docs to reflect this new behavior where users no longer need to set task.inputs for low level application.

mynameborat · 2019-02-25T18:01:20Z

samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala

@@ -358,14 +369,6 @@ object SamzaContainer extends Logging {

    info("Got intermediate streams: %s" format intermediateStreams)

-    val sideInputStoresToSystemStreams = config.getStoreNames


Not sure if this is a git diff thing. I notice some usages of sideInputStoresToSystemStreams below which are not removed as part of this PR. e.g. ln 546, 567
Can you confirm?

i just moved it up

mynameborat · 2019-02-25T18:13:16Z

samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala

@@ -656,13 +593,11 @@ object SamzaContainer extends Logging {
          storageManager = storageManager,
          tableManager = tableManager,
          reporters = reporters,
-          systemStreamPartitions = taskSSPs,
+          systemStreamPartitions = taskSSPs -- taskSideInputSSPs,


Are we piggy backing on ssps within task model to pass the side inputs too? Should we have something similar to changelog partition in task model that represents the side inputs separately so that we don't have to handle for side inputs in places where taskSSPs is used?

Lets do it separately under https://issues.apache.org/jira/browse/SAMZA-2112

mynameborat · 2019-02-25T18:50:23Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

+      // Make the main thread wait until all sideInputs have been caughtup or thrown an exception
+      this.sideInputsCaughtUp.await();
+
+      if (sideInputException.isPresent()) { // Throw exception if there was an exception in catching-up sideInputs


Can you add the jira ticket if exists to capture the scenario where we need to communicate to the container in case of the exception after bootstrapping?

mynameborat · 2019-02-25T18:51:55Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

+    sideInputsFlushFuture = sideInputsFlushExecutor.scheduleWithFixedDelay(new Runnable() {
+      @Override
+      public void run() {
+        getSideInputStorageManagers().forEach(sideInputStorageManager -> sideInputStorageManager.flush());


what happens when flush throws an exception? Can we also capture that into sideInputException?

mynameborat · 2019-02-25T18:53:11Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

+    try {
+      sideInputsFlushExecutor.awaitTermination(SIDE_INPUT_FLUSH_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+    } catch (InterruptedException e) {
+      throw new SamzaException("Exception while shutting down side inputs", e);


We also need to bubble up the sideInputException if set here.

mynameborat · 2019-02-25T18:56:57Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

-            storageEngine.stop();
-          });
+
+      Map<String, StorageEngine> persistentStores = this.taskStores.entrySet().stream().filter(e -> {


I noticed you use taskStores from the outer scope from CSM.
Can we instead use the helper method you have from outer class to get non-side-input stores to be safe?

mynameborat · 2019-02-25T18:59:48Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

+
+    // stop all sideinput consumers and stores
+    if (sideInputSystemConsumers != null) {
+      this.sideInputSystemConsumers.stop();


Should this go higher before shutting down the read? Wondering if the consumers have buffer which also needs to emptied before stopping the read.

Along the same lines as SamzaContainer, which first shuts down runloop, then stops consumers

rmatharu-zz · 2019-02-26T01:03:45Z

Addressed all comments.

mynameborat

Looks good overall. Considered approved once you have addressed the comments.

mynameborat · 2019-02-27T22:09:38Z

samza-core/src/main/java/org/apache/samza/storage/TaskSideInputStorageManager.java

@@ -239,7 +250,7 @@ private void initializeStoreDirectories() {
   * Its contents are a JSON encoded mapping from each side input SSP to its last processed offset, and a checksum.
   */
  @VisibleForTesting
-  void writeOffsetFiles() {
+  void  writeOffsetFiles() {


nit: revert extra spacing

mynameborat · 2019-02-27T22:26:15Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

+  private volatile CountDownLatch sideInputsCaughtUp; // Used by the sideInput-read thread to signal to the main thread
+  private volatile boolean shutDownSideInputRead = false;
+  private final ScheduledExecutorService sideInputsFlushExecutor = Executors.newSingleThreadScheduledExecutor(
+      new ThreadFactoryBuilder().setNameFormat(SIDEINPUTS_FLUSH_THREAD_NAME).build());


Can you set the thread to be daemon?

mynameborat · 2019-02-27T22:27:59Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

+    }
+
+    // stop all sideInputStores -- this will perform one last flush on the KV stores, and write the offset file
+    this.sideInputStorageManagers.values().stream().collect(Collectors.toSet()).


nit: reuse getSideInputStorageManagers()

mynameborat · 2019-02-27T22:29:35Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

+    this.sideInputSystemConsumers.start();
+
+    // create a thread for sideInput reads
+    Thread readSideInputs = new Thread(() -> {


Can we set this to be a daemon thread? The java default is non-daemon thread.

mynameborat · 2019-02-27T22:34:27Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java


    // creating task restore managers
    this.taskRestoreManagers = createTaskRestoreManagers(systemAdmins, clock);
+
+    // creating side input store processors, one per store per task
+    taskSideInputProcessors = createSideInputProcessors(new StorageConfig(config), this.containerModel, this.sideInputSystemStreams, this.taskInstanceMetrics);


Looks like this can be a local variable or done inside the createSideInputStorageManagers()

mynameborat · 2019-02-27T22:41:10Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

+    this.shutDownSideInputRead = true;
+
+    // stop all sideinput consumers and stores
+    if (sideInputSystemConsumers != null) {


Can we simplify these nulls checks and instead have one state that denotes if side inputs are present or absent? I noticed in some places you check if the sideInputConsumers are non-empty and some of them checks for null and some use sideInputSystemStreams is non-empty.

Lets just have one place and one logic to determine the presence of side inputs and if its present then the invariant of some of the variables being non-null should hold since you create empty maps or collection for the most part.

mynameborat · 2019-02-27T22:44:46Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

+    this.sideInputConsumers = createConsumers(this.sideInputSystemStreams, systemFactories, config, this.samzaContainerMetrics.registry());
+
+    // create SystemConsumers for consuming from taskSideInputSSPs, if sideInputs are being used
+    if (!this.sideInputConsumers.isEmpty()) {


please refer to comment below about the checks for presence of side inputs

mynameborat · 2019-02-27T22:49:46Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

-            .get(storeName)
-            .getStoreProperties()
-            .isPersistedToDisk()) {
+        // if this store has been already then re-create and overwrite it only if it is a


nit: missing "created in the taskStores"

mynameborat · 2019-02-27T22:52:13Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

+          // have to use the right serde because the sideInput stores are created
+          Serde keySerde = serdes.get(new StorageConfig(config).getStorageKeySerde(storeName).get());
+          Serde msgSerde = serdes.get(new StorageConfig(config).getStorageMsgSerde(storeName).get());
+          sideInputStoresToProcessors.get(taskName).put(storeName, new SideInputsProcessor() {


nit: extract into constant since its okay to reuse the same identity processor for all stores.

Each store is created with its respective configured serdes, so this processor needs to use that serde.

mynameborat · 2019-02-27T22:53:43Z

samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java

+
+    containerModel.getTasks().forEach((taskName, taskModel) -> {
+
+        if (!sideInputSystemStreams.isEmpty()) {


can you move it outside the forEach to short circuit earlier?

rmatharu-zz · 2019-02-28T02:25:04Z

Addressed all comments.

…geManager (apache#912) * Rocksdb bug fix * Moving TaskSideInputStorageManagers in CSM * Adding basic side input refactor. Consumption logic works from inside CSM. * Cleaning up sideInput references from TaskInstances and SamzaContainer * minor * Adding periodic flush logic to CSM for sideInputs * Removing sideInputs from taskInputs * fixing test * test fix * test fix * Code simplifications * checkstyle * Using sideInput storage manager for standby tasks * Minor store dir integ * Adding comment * Adding merge function in case of duplicate keys * fixing tests * Addressing cosmetic review comments * Added bound-flush and exception handling for sideInputs * Adding documentation * TestTaskStorageManager : updating test * Adding taskMode to decide StorePartitionDir * Adding taskMode to decide StorePartitionDir minor * minor * minor refactor of CSM * minor * Addressing review comments * minor change in CSM

rmatharu added 15 commits January 23, 2019 18:16

Rocksdb bug fix

2c09b08

Merge branch 'master' of https://github.com/apache/samza

916f66a

Merge branch 'master' of https://github.com/apache/samza

0800687

Merge branch 'master' of https://github.com/apache/samza

ee7daac

Merge branch 'master' of https://github.com/apache/samza

12ca96b

Moving TaskSideInputStorageManagers in CSM

f8ac4f2

Adding basic side input refactor. Consumption logic works from inside…

3e0e5d8

… CSM.

Cleaning up sideInput references from TaskInstances and SamzaContainer

76c9cd2

minor

6c62223

Adding periodic flush logic to CSM for sideInputs

05a1b02

Removing sideInputs from taskInputs

23ab4f8

fixing test

8e4b648

test fix

574eedd

test fix

cd7779a

Code simplifications

bb675f7

rmatharu-zz changed the title ~~Test sideinputrefactor~~ Refactoring sideInputs from SamzaContainer to ContainerStorageManager Feb 8, 2019

checkstyle

190bc7c

rmatharu-zz mentioned this pull request Feb 8, 2019

SEP-19 : Basic Standby task/container implementation #876

Closed

rmatharu-zz changed the title ~~Refactoring sideInputs from SamzaContainer to ContainerStorageManager~~ SEP-19 : Refactoring sideInputs from SamzaContainer to ContainerStorageManager Feb 8, 2019

rmatharu added 10 commits February 8, 2019 21:20

Using sideInput storage manager for standby tasks

a0585e1

Minor store dir integ

9608743

Merge branch 'test-sideinputrefactor' into test-standbytask-with-side…

d396373

…input

Adding comment

c993f0f

Adding merge function in case of duplicate keys

7b2117b

fixing tests

d802e59

Merge branch 'master' of https://github.com/apache/samza

36c0b33

Merge branch 'master' of https://github.com/apache/samza

1a72dc4

Merge branch 'master' of https://github.com/apache/samza

497602a

Merge branch 'master' of https://github.com/apache/samza

40f68a6

mynameborat requested changes Feb 19, 2019

View reviewed changes

rmatharu-zz and others added 7 commits February 19, 2019 20:12

Merge branch 'master' into test-sideinputrefactor

7ca1d04

Addressing cosmetic review comments

6f6fc42

Added bound-flush and exception handling for sideInputs

4953e59

Merge branch 'test-sideinputrefactor' of https://github.com/rmatharu/…

5a94299

…samza into test-sideinputrefactor

Merge branch 'master' of https://github.com/apache/samza

96e3d8f

Merge branch 'master' into test-sideinputrefactor

908af23

Adding documentation

e006acd

rmatharu added 5 commits February 21, 2019 20:37

TestTaskStorageManager : updating test

a96e5ef

Adding taskMode to decide StorePartitionDir

e9c05dc

Adding taskMode to decide StorePartitionDir

aeedc00

minor

Merge branch 'test-sideinputrefactor' of https://github.com/rmatharu/…

a5872fc

…samza into test-sideinputrefactor

minor

cfda86a

mynameborat requested changes Feb 25, 2019

View reviewed changes

rmatharu added 2 commits February 25, 2019 16:07

minor refactor of CSM

c65845b

minor

0538e62

mynameborat requested changes Feb 27, 2019

View reviewed changes

Addressing review comments

8bba0e6

mynameborat approved these changes Feb 28, 2019

View reviewed changes

minor change in CSM

9dfd12b

atoomula merged commit 320359f into apache:master Mar 1, 2019

		@@ -112,15 +144,49 @@

		public ContainerStorageManager(ContainerModel containerModel, StreamMetadataCache streamMetadataCache,

		@@ -358,14 +369,6 @@ object SamzaContainer extends Logging {

		info("Got intermediate streams: %s" format intermediateStreams)

		val sideInputStoresToSystemStreams = config.getStoreNames


		containerModel.getTasks().forEach((taskName, taskModel) -> {

		if (!sideInputSystemStreams.isEmpty()) {

SEP-19 : Refactoring sideInputs from SamzaContainer to ContainerStorageManager #912

SEP-19 : Refactoring sideInputs from SamzaContainer to ContainerStorageManager #912

Conversation

rmatharu-zz commented Feb 8, 2019 • edited

mynameborat left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rmatharu-zz Feb 22, 2019 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rmatharu-zz commented Feb 22, 2019

mynameborat left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rmatharu-zz commented Feb 26, 2019

mynameborat left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rmatharu-zz commented Feb 28, 2019

rmatharu-zz commented Feb 8, 2019 •

edited

rmatharu-zz Feb 22, 2019 •

edited