private static Function<OperatorSpec, Iterable<OperatorSpec>> getCustomGetNextOpSpecs(
      Iterable<InputOperatorSpec> inputOpSpecs) {

    // Traverse operatorSpecGraph to create mapping between every SendToTableOperatorSpec and all
    // StreamTableJoinOperatorSpecs referencing the same table.
    TableJoinVisitor tableJoinVisitor = new TableJoinVisitor();
    for (InputOperatorSpec inputOpSpec : inputOpSpecs) {
      traverse(inputOpSpec, tableJoinVisitor, opSpec -> opSpec.getRegisteredOperatorSpecs());
    }

    Multimap<SendToTableOperatorSpec, StreamTableJoinOperatorSpec> sendToTableOpSpecToStreamTableJoinOpSpecs =
        tableJoinVisitor.getSendToTableOpSpecToStreamTableJoinOpSpecs();

    return operatorSpec -> {
      // If this is a SendToTableOperatorSpec, return all StreamTableJoinSpecs referencing the same table.
      // For all other types of operator specs, return the next registered operator specs.
      if (operatorSpec instanceof SendToTableOperatorSpec) {
        SendToTableOperatorSpec sendToTableOperatorSpec = (SendToTableOperatorSpec) operatorSpec;
        return Collections.unmodifiableCollection(sendToTableOpSpecToStreamTableJoinOpSpecs.get(sendToTableOperatorSpec));
      }

      return operatorSpec.getRegisteredOperatorSpecs();
    };
  }
Creates a function that retrieves the next {@link OperatorSpec}s of any given {@link OperatorSpec} in the specified
{@code operatorSpecGraph}.

Calling the returned function with any {@link SendToTableOperatorSpec} will return a collection of all
{@link StreamTableJoinOperatorSpec}s that reference the same table as the specified
{@link SendToTableOperatorSpec}, as if they were actually connected.
samza-core/src/main/java/org/apache/samza/execution/OperatorSpecGraphAnalyzer.java