public Map<TaskName, Map<SystemStreamPartition, Startpoint>> fanOut(Map<TaskName, Set<SystemStreamPartition>> taskToSSPs) throws IOException {
    Preconditions.checkState(!stopped, "Underlying metadata store not available");
    Preconditions.checkArgument(MapUtils.isNotEmpty(taskToSSPs), "taskToSSPs cannot be null or empty");

    // construct fan out with the existing readWriteStore entries and mark the entries for deletion after fan out
    Instant now = Instant.now();
    HashMultimap<SystemStreamPartition, TaskName> deleteKeys = HashMultimap.create();
    HashMap<TaskName, StartpointFanOutPerTask> fanOuts = new HashMap<>();
    Map<String, byte[]> startpointMap = readWriteStore.all();

    for (TaskName taskName : taskToSSPs.keySet()) {
      Set<SystemStreamPartition> ssps = taskToSSPs.get(taskName);
      if (CollectionUtils.isEmpty(ssps)) {
        LOG.warn("No SSPs are mapped to taskName: {}", taskName.getTaskName());
        continue;
      }
      for (SystemStreamPartition ssp : ssps) {
        Optional<Startpoint> startpoint = readStartpoint(startpointMap, ssp, null); // Read SSP-only key
        startpoint.ifPresent(sp -> deleteKeys.put(ssp, null));

        Optional<Startpoint> startpointForTask = readStartpoint(startpointMap, ssp, taskName); // Read SSP+taskName key
        startpointForTask.ifPresent(sp -> deleteKeys.put(ssp, taskName));

        Optional<Startpoint> startpointWithPrecedence = resolveStartpointPrecendence(startpoint, startpointForTask);
        if (!startpointWithPrecedence.isPresent()) {
          continue;
        }

        fanOuts.putIfAbsent(taskName, new StartpointFanOutPerTask(now));
        fanOuts.get(taskName).getFanOuts().put(ssp, startpointWithPrecedence.get());
      }
    }

    if (fanOuts.isEmpty()) {
      LOG.debug("No fan outs created.");
      return ImmutableMap.of();
    }

    LOG.info("Fanning out to {} tasks", fanOuts.size());

    // Fan out to store
    for (TaskName taskName : fanOuts.keySet()) {
      String fanOutKey = toFanOutStoreKey(taskName);
      StartpointFanOutPerTask newFanOut = fanOuts.get(taskName);
      fanOutStore.put(fanOutKey, objectMapper.writeValueAsBytes(newFanOut));
    }
    fanOutStore.flush();

    for (SystemStreamPartition ssp : deleteKeys.keySet()) {
      for (TaskName taskName : deleteKeys.get(ssp)) {
        if (taskName != null) {
          deleteStartpoint(ssp, taskName);
        } else {
          deleteStartpoint(ssp);
        }
      }
    }

    return ImmutableMap.copyOf(fanOuts.entrySet().stream()
        .collect(Collectors.toMap(fo -> fo.getKey(), fo -> fo.getValue().getFanOuts())));
  }
The Startpoints that are written to with {@link #writeStartpoint(SystemStreamPartition, Startpoint)} and with
{@link #writeStartpoint(SystemStreamPartition, TaskName, Startpoint)} are moved from a "read-write" namespace
to a "fan out" namespace.
This method is not atomic nor thread-safe. The intent is for the Samza Processor's coordinator to use this
method to assign the Startpoints to the appropriate tasks.
@param taskToSSPs Determines which {@link TaskName} each {@link SystemStreamPartition} maps to.
@return The set of active {@link TaskName}s that were fanned out to.
samza-core/src/main/java/org/apache/samza/startpoint/StartpointManager.java