@VisibleForTesting
  Map<String, Object> operatorToMap(OperatorSpec spec) {
    Map<String, Object> map = new HashMap<>();
    map.put("opCode", spec.getOpCode().name());
    map.put("opId", spec.getOpId());
    map.put("sourceLocation", spec.getSourceLocation());

    Collection<OperatorSpec> nextOperators = spec.getRegisteredOperatorSpecs();
    map.put("nextOperatorIds", nextOperators.stream().map(OperatorSpec::getOpId).collect(Collectors.toSet()));

    if (spec instanceof OutputOperatorSpec) {
      OutputStreamImpl outputStream = ((OutputOperatorSpec) spec).getOutputStream();
      map.put("outputStreamId", outputStream.getStreamId());
    } else if (spec instanceof PartitionByOperatorSpec) {
      OutputStreamImpl outputStream = ((PartitionByOperatorSpec) spec).getOutputStream();
      map.put("outputStreamId", outputStream.getStreamId());
    }

    if (spec instanceof StreamTableJoinOperatorSpec) {
      String tableId = ((StreamTableJoinOperatorSpec) spec).getTableId();
      map.put("tableId", tableId);
    }

    if (spec instanceof SendToTableOperatorSpec) {
      String tableId = ((SendToTableOperatorSpec) spec).getTableId();
      map.put("tableId", tableId);
    }

    if (spec instanceof JoinOperatorSpec) {
      map.put("ttlMs", ((JoinOperatorSpec) spec).getTtlMs());
    }

    return map;
  }
Format the operator properties into a map
@param spec a {@link OperatorSpec} instance
@return map of the operator properties
samza-core/src/main/java/org/apache/samza/execution/JobGraphJsonGenerator.java