rmatharu 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_r261025331
########## File path: samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java ########## @@ -330,14 +486,110 @@ private StorageEngine createStore(String storeName, TaskName taskName, TaskModel storeMetricsRegistry, changeLogSystemStreamPartition, jobContext, containerContext, storeMode); } + + // Create side input store processors, one per store per task + private Map<TaskName, Map<String, SideInputsProcessor>> createSideInputProcessors(StorageConfig config, ContainerModel containerModel, + Map<String, Set<SystemStream>> sideInputSystemStreams, Map<TaskName, TaskInstanceMetrics> taskInstanceMetrics) { + + Map<TaskName, Map<String, SideInputsProcessor>> sideInputStoresToProcessors = new HashMap<>(); + getTasks(containerModel, TaskMode.Active).forEach((taskName, taskModel) -> { + sideInputStoresToProcessors.put(taskName, new HashMap<>()); + for (String storeName : sideInputSystemStreams.keySet()) { + if (config.getSideInputsProcessorSerializedInstance(storeName).isDefined()) { + sideInputStoresToProcessors.get(taskName) + .put(storeName, SerdeUtils.deserialize("Side Inputs Processor", + config.getSideInputsProcessorSerializedInstance(storeName).get())); + } else { + sideInputStoresToProcessors.get(taskName) + .put(storeName, Util.getObj(config.getSideInputsProcessorFactory(storeName).get(), + SideInputsProcessorFactory.class).getSideInputsProcessor(config, + taskInstanceMetrics.get(taskName).registry())); + } + } + }); + + // creating identity sideInputProcessor for stores of standbyTasks + getTasks(containerModel, TaskMode.Standby).forEach((taskName, taskModel) -> { + sideInputStoresToProcessors.put(taskName, new HashMap<>()); + for (String storeName : sideInputSystemStreams.keySet()) { + + // 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() { Review comment: Each store is created with its respective configured serdes, so this processor needs to use that serde. ---------------------------------------------------------------- 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