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_r260980813
########## 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() { + @Override + public Collection<Entry<?, ?>> process(IncomingMessageEnvelope message, KeyValueStore store) { + return ImmutableList.of(new Entry<>(keySerde.fromBytes((byte[]) message.getKey()), msgSerde.fromBytes((byte[]) message.getMessage()))); + } + }); + } + }); + + return sideInputStoresToProcessors; + } + + // Create task side input storage managers, one per task, index by the SSP they are responsible for consuming + private Map<SystemStreamPartition, TaskSideInputStorageManager> createSideInputStorageManagers(Clock clock) { + + Map<SystemStreamPartition, TaskSideInputStorageManager> sideInputStorageManagers = new HashMap<>(); + + containerModel.getTasks().forEach((taskName, taskModel) -> { + + if (!sideInputSystemStreams.isEmpty()) { Review comment: can you move it outside the forEach to short circuit earlier? ---------------------------------------------------------------- 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