code: @Override
    public Table<Node, String, Set<QueueSpecification>> create(FlowSpecification input) {
      Table<Node, String, Set<QueueSpecification>> table = HashBasedTable.create();
  
      String flow = input.getName();
      Map<String, FlowletDefinition> flowlets = input.getFlowlets();
  
      // Iterate through connections of a flow.
      for (FlowletConnection connection : input.getConnections()) {
        final String source = connection.getSourceName();
        final String target = connection.getTargetName();
        final Node sourceNode = new Node(connection.getSourceType(), source);
  
        Set<QueueSpecification> queueSpec = generateQueueSpecification(flow, connection, flowlets.get(target).getInputs(),
                                                                       flowlets.get(source).getOutputs());
  
        Set<QueueSpecification> queueSpecifications = table.get(sourceNode, target);
        if (queueSpecifications == null) {
          queueSpecifications = Sets.newHashSet();
          table.put(sourceNode, target, queueSpecifications);
        }
        queueSpecifications.addAll(queueSpec);
      }
      return table;
    }
docs: Given a {@link FlowSpecification}.
  
  @param input {@link FlowSpecification}
  @return A {@link com.google.common.collect.Table}