code: protected Set<QueueSpecification> generateQueueSpecification(String flow,
                                                                 FlowletConnection connection,
                                                                 Map<String, Set<Schema>> inputSchemas,
                                                                 Map<String, Set<Schema>> outputSchemas) {
  
      ImmutableSet.Builder<QueueSpecification> builder = ImmutableSet.builder();
  
      // Iterate through all the outputs and look for an input with compatible schema.
      // If no such input is found, look for one with ANY_INPUT.
      // If there is exact match of schema then we pick that over the compatible one.
      for (Map.Entry<String, Set<Schema>> entryOutput : outputSchemas.entrySet()) {
        String outputName = entryOutput.getKey();
  
        Set<Schema> nameOutputSchemas = inputSchemas.get(outputName);
        ImmutablePair<Schema, Schema> schemas =
          (nameOutputSchemas == null) ? null : SchemaFinder.findSchema(entryOutput.getValue(), nameOutputSchemas);
  
        if (schemas == null) {
          // Try ANY_INPUT
          Set<Schema> anyInputSchemas = inputSchemas.get(FlowletDefinition.ANY_INPUT);
          if (anyInputSchemas != null) {
            schemas = SchemaFinder.findSchema(entryOutput.getValue(), anyInputSchemas);
          }
        }
  
        if (schemas == null) {
          continue;
        }
  
        if (connection.getSourceType() == FlowletConnection.Type.STREAM) {
          builder.add(createSpec(QueueName.fromStream(outputName),
                                 schemas.getFirst(), schemas.getSecond()));
        } else {
          builder.add(createSpec(QueueName.fromFlowlet(flow, flow,
                                                       connection.getSourceName(), outputName),
                                 schemas.getFirst(), schemas.getSecond()));
        }
      }
  
      return builder.build();
    }
docs: Finds a equal or compatible schema connection between <code>source</code> and <code>target</code>
  flowlet.