private static Map<SystemStreamPartition, TaskSideInputHandler> createSideInputHandlers(
      boolean hasSideInputs, Map<TaskName, Map<String, StorageEngine>> sideInputStores,
      Map<TaskName, Map<String, Set<SystemStreamPartition>>> taskSideInputStoreSSPs,
      Map<TaskName, CountDownLatch> sideInputTaskLatches, Map<TaskName, TaskInstanceMetrics> taskInstanceMetrics,
      ContainerModel containerModel, StreamMetadataCache streamMetadataCache, SystemAdmins systemAdmins,
      Map<String, Serde<Object>> serdes, File loggedStoreBaseDirectory, File nonLoggedStoreBaseDirectory,
      Config config, Clock clock) {
    // creating sideInput store processors, one per store per task
    Map<TaskName, Map<String, SideInputsProcessor>> taskSideInputProcessors =
        createSideInputProcessors(taskInstanceMetrics, taskSideInputStoreSSPs,
            containerModel, serdes, new StorageConfig(config));

    Map<SystemStreamPartition, TaskSideInputHandler> handlers = new HashMap<>();

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

        Map<String, StorageEngine> taskSideInputStores = sideInputStores.get(taskName);
        Map<String, Set<SystemStreamPartition>> sideInputStoresToSSPs = new HashMap<>();
        boolean taskHasSideInputs = false;
        for (String storeName : taskSideInputStores.keySet()) {
          Set<SystemStreamPartition> storeSSPs = taskSideInputStoreSSPs.get(taskName).get(storeName);
          taskHasSideInputs = taskHasSideInputs || !storeSSPs.isEmpty();
          sideInputStoresToSSPs.put(storeName, storeSSPs);
        }

        if (taskHasSideInputs) {
          CountDownLatch taskCountDownLatch = new CountDownLatch(1);
          sideInputTaskLatches.put(taskName, taskCountDownLatch);

          TaskSideInputHandler taskSideInputHandler = new TaskSideInputHandler(taskName,
              taskModel.getTaskMode(),
              loggedStoreBaseDirectory,
              taskSideInputStores,
              sideInputStoresToSSPs,
              taskSideInputProcessors.get(taskName),
              systemAdmins,
              streamMetadataCache,
              taskCountDownLatch,
              clock);

          sideInputStoresToSSPs.values().stream().flatMap(Set::stream).forEach(ssp -> {
            handlers.put(ssp, taskSideInputHandler);
          });

          LOG.info("Created TaskSideInputHandler for task {}, taskSideInputStores {} and loggedStoreBaseDirectory {}",
              taskName, taskSideInputStores, loggedStoreBaseDirectory);
        }
      });
    }
    return handlers;
  }
Add all sideInputs to a map of maps, indexed first by taskName, then by sideInput store name.

@param sideInputSystemStreams the map of store to sideInput system stream
@param changelogSystemStreams the map of store to changelog system stream
@param containerModel the containerModel to use
@return taskSideInputSSPs map
samza-core/src/main/scala/org/apache/samza/storage/SideInputsManager.java