A factory to generate {@link WorkUnitStream} given a generic type datasetIterator.
@throws IOException
16284
protected Iterator<WorkUnit> getWorkUnitIterator(Iterator<Dataset> datasetIterator, String previousDatasetUrnWatermark,
      @Nullable String previousPartitionUrnWatermark, int maximumWorkUnits) throws IOException {
    return new DeepIterator(datasetIterator, previousDatasetUrnWatermark, previousPartitionUrnWatermark,
            maximumWorkUnits);
  }
573624
public void generate( IWTypeLib lib ) throws BindingException, IOException {
        LibBinder tli = getTypeLibInfo(lib);
        if(referenceResolver.suppress(lib))
            return; // skip code generation
        if(generatedTypeLibs.add(tli))
            tli.generate();
    }
1122387
@SafeVarargs
    @SuppressWarnings("unchecked")
    public static <T> Stream<T> stream(Iterable<T>... iterables) {
        List<Iterator<T>> iterators = Arrays.stream(iterables)
            .map(Iterable::iterator)
            .collect(Collectors.toList());

        return stream(iterators.toArray(new Iterator[iterables.length]));
    }
1081686
@Override
  public List<Dataset> findDatasets() throws IOException {
    Set<URI> leafDatasets = getValidDatasetURIs(this.commonRoot);
    if (leafDatasets.isEmpty()) {
      return ImmutableList.of();
    }

    // Parallel execution for copyDataset for performance consideration.
    final List<Dataset> result = new CopyOnWriteArrayList<>();
    Iterator<Callable<Void>> callableIterator =
        Iterators.transform(leafDatasets.iterator(), new Function<URI, Callable<Void>>() {
          @Override
          public Callable<Void> apply(final URI datasetURI) {
            return findDatasetsCallable(configClient, datasetURI, props, blacklistPatterns, result);
          }
        });

    this.executeItertorExecutor(callableIterator);
    log.info("found {} datasets in ConfigBasedDatasetsFinder", result.size());
    return result;
  }
1169099
public static ByteIterator of(final Supplier<? extends ByteIterator> iteratorSupplier) {
        N.checkArgNotNull(iteratorSupplier, "iteratorSupplier");

        return new ByteIterator() {
            private ByteIterator iter = null;
            private boolean isInitialized = false;

            @Override
            public boolean hasNext() {
                if (isInitialized == false) {
                    init();
                }

                return iter.hasNext();
            }

            @Override
            public byte nextByte() {
                if (isInitialized == false) {
                    init();
                }

                return iter.nextByte();
            }

            private void init() {
                if (isInitialized == false) {
                    isInitialized = true;
                    iter = iteratorSupplier.get();
                }
            }
        };
    }
890068
@Override
    public Object generate(SequenceGeneratorDiscriptor discriptor, Client<?> client, String dataType)
    {
        Jedis jedis = ((RedisClient) client).factory.getConnection();

        Long latestCount = jedis.incr(((RedisClient) client).getEncodedBytes(discriptor.getSequenceName()));
        if (latestCount == 1)
        {
            return discriptor.getInitialValue();
        }
        else
        {
            return (latestCount - 1) * discriptor.getAllocationSize();
        }
    }
184570
@Override
	public DataSet<Tuple2<VertexKey, VertexValue>> createResult() {
		if (this.initialVertices == null) {
			throw new IllegalStateException("The input data set has not been set.");
		}
		
		// prepare some type information
		TypeInformation<Tuple2<VertexKey, VertexValue>> vertexTypes = initialVertices.getType();
		TypeInformation<VertexKey> keyType = ((TupleTypeInfo<?>) initialVertices.getType()).getTypeAt(0);
		TypeInformation<Tuple2<VertexKey, Message>> messageTypeInfo = new TupleTypeInfo<Tuple2<VertexKey,Message>>(keyType, messageType);		
		
		// set up the iteration operator
		final String name = (this.name != null) ? this.name :
			"Vertex-centric iteration (" + updateFunction + " | " + messagingFunction + ")";
		final int[] zeroKeyPos = new int[] {0};
	
		final DeltaIteration<Tuple2<VertexKey, VertexValue>, Tuple2<VertexKey, VertexValue>> iteration =
			this.initialVertices.iterateDelta(this.initialVertices, this.maximumNumberOfIterations, zeroKeyPos);
		iteration.name(name);
		iteration.parallelism(parallelism);
		
		// register all aggregators
		for (Map.Entry<String, Class<? extends Aggregator<?>>> entry : this.aggregators.entrySet()) {
			iteration.registerAggregator(entry.getKey(), entry.getValue());
		}
		
		// build the messaging function (co group)
		CoGroupOperator<?, ?, Tuple2<VertexKey, Message>> messages;
		if (edgesWithoutValue != null) {
			MessagingUdfNoEdgeValues<VertexKey, VertexValue, Message> messenger = new MessagingUdfNoEdgeValues<VertexKey, VertexValue, Message>(messagingFunction, messageTypeInfo);
			messages = this.edgesWithoutValue.coGroup(iteration.getWorkset()).where(0).equalTo(0).with(messenger);
		}
		else {
			MessagingUdfWithEdgeValues<VertexKey, VertexValue, Message, EdgeValue> messenger = new MessagingUdfWithEdgeValues<VertexKey, VertexValue, Message, EdgeValue>(messagingFunction, messageTypeInfo);
			messages = this.edgesWithValue.coGroup(iteration.getWorkset()).where(0).equalTo(0).with(messenger);
		}
		
		// configure coGroup message function with name and broadcast variables
		messages = messages.name("Messaging");
		for (Tuple2<String, DataSet<?>> e : this.bcVarsMessaging) {
			messages = messages.withBroadcastSet(e.f1, e.f0);
		}
		
		VertexUpdateUdf<VertexKey, VertexValue, Message> updateUdf = new VertexUpdateUdf<VertexKey, VertexValue, Message>(updateFunction, vertexTypes);
		
		// build the update function (co group)
		CoGroupOperator<?, ?, Tuple2<VertexKey, VertexValue>> updates =
				messages.coGroup(iteration.getSolutionSet()).where(0).equalTo(0).with(updateUdf);
		
		// configure coGroup update function with name and broadcast variables
		updates = updates.name("Vertex State Updates");
		for (Tuple2<String, DataSet<?>> e : this.bcVarsUpdate) {
			updates = updates.withBroadcastSet(e.f1, e.f0);
		}

		// let the operator know that we preserve the key field
		updates.withConstantSetFirst("0").withConstantSetSecond("0");
		
		return iteration.closeWith(updates, updates);
		
	}
734116
public static void writeGenerators( IGeneratorSet generators, OutputStream outputStream)
    {
    try( GeneratorSetJsonWriter writer = new GeneratorSetJsonWriter( outputStream))
      {
      writer.write( generators);
      }
    catch( Exception e)
      {
      throw new RuntimeException( "Can't write generator definitions", e);
      }
    }
844571
public static <T> Collector<T, List<T>> toList() {
    return new Collector<T, List<T>>() {
      @Override
      public List<T> collect(Stream<? extends T> stream) {
        return Lists.newArrayList(stream.iterator());
      }
    };
  }
695141
public <T0, T1, T2> SingleOutputStreamOperator<Tuple3<T0, T1, T2>> projectTuple3() {
		TypeInformation<?>[] fTypes = extractFieldTypes(fieldIndexes, dataStream.getType());
		TupleTypeInfo<Tuple3<T0, T1, T2>> tType = new TupleTypeInfo<Tuple3<T0, T1, T2>>(fTypes);

		return dataStream.transform("Projection", tType, new StreamProject<IN, Tuple3<T0, T1, T2>>(fieldIndexes, tType.createSerializer(dataStream.getExecutionConfig())));
	}
778818
private List<WorkUnit> materializeWorkUnitList(WorkUnitStream workUnitStream) {
    if (!workUnitStream.isFiniteStream()) {
      throw new UnsupportedOperationException("Cannot materialize an infinite work unit stream.");
    }
    return Lists.newArrayList(workUnitStream.getWorkUnits());
  }
1192314
public <T0, T1, T2, T3, T4, T5> SingleOutputStreamOperator<Tuple6<T0, T1, T2, T3, T4, T5>> projectTuple6() {
		TypeInformation<?>[] fTypes = extractFieldTypes(fieldIndexes, dataStream.getType());
		TupleTypeInfo<Tuple6<T0, T1, T2, T3, T4, T5>> tType = new TupleTypeInfo<Tuple6<T0, T1, T2, T3, T4, T5>>(fTypes);

		return dataStream.transform("Projection", tType, new StreamProject<IN, Tuple6<T0, T1, T2, T3, T4, T5>>(fieldIndexes, tType.createSerializer(dataStream.getExecutionConfig())));
	}
1295073
public static void write(IDataSet dataSet, OutputStream out) throws IOException, DataSetException {
		logger.debug("write(dataSet={}, out={}) - start", dataSet, out);

		FlatXmlWriter datasetWriter = new FlatXmlWriter(out);
		datasetWriter.setIncludeEmptyTable(true);
		datasetWriter.write(dataSet);
	}
1203184
public <T> T read(final InputStream stream, final DataTypeDescriptor<T> descriptor) {
		try {
			return jsonFormat.read(stream, descriptor);
		} catch (Exception e) {
			throw propagate(e);
		}
	}
1215592
public static <T> Stream<T> toIterator(@javax.annotation.Nonnull final Iterator<T> iterator) {
    return StreamSupport.stream(Spliterators.spliterator(iterator, 1, Spliterator.ORDERED), false);
  }
797835
private JobConfig.Builder createJob(List<WorkUnit> workUnits) throws IOException {
    Map<String, TaskConfig> taskConfigMap = Maps.newHashMap();

    try (ParallelRunner stateSerDeRunner = new ParallelRunner(this.stateSerDeRunnerThreads, this.fs)) {
      int multiTaskIdSequence = 0;
      for (WorkUnit workUnit : workUnits) {
        if (workUnit instanceof MultiWorkUnit) {
          workUnit.setId(JobLauncherUtils.newMultiTaskId(this.jobContext.getJobId(), multiTaskIdSequence++));
        }
        addWorkUnit(workUnit, stateSerDeRunner, taskConfigMap);
      }

      Path jobStateFilePath;

      // write the job.state using the state store if present, otherwise serialize directly to the file
      if (this.stateStores.haveJobStateStore()) {
        jobStateFilePath = GobblinClusterUtils.getJobStateFilePath(true, this.appWorkDir, this.jobContext.getJobId());
        this.stateStores.getJobStateStore().put(jobStateFilePath.getParent().getName(), jobStateFilePath.getName(),
            this.jobContext.getJobState());
      } else {
        jobStateFilePath = GobblinClusterUtils.getJobStateFilePath(false, this.appWorkDir, this.jobContext.getJobId());
        SerializationUtils.serializeState(this.fs, jobStateFilePath, this.jobContext.getJobState());
      }

      LOGGER.debug("GobblinHelixJobLauncher.createJob: jobStateFilePath {}, jobState {} jobProperties {}",
          jobStateFilePath, this.jobContext.getJobState().toString(), this.jobContext.getJobState().getProperties());
    }

    JobConfig.Builder jobConfigBuilder = new JobConfig.Builder();

    // Helix task attempts = retries + 1 (fallback to general task retry for backward compatibility)
    jobConfigBuilder.setMaxAttemptsPerTask(this.jobContext.getJobState().getPropAsInt(
        GobblinClusterConfigurationKeys.HELIX_TASK_MAX_ATTEMPTS_KEY,
        this.jobContext.getJobState().getPropAsInt(
            ConfigurationKeys.MAX_TASK_RETRIES_KEY,
            ConfigurationKeys.DEFAULT_MAX_TASK_RETRIES)) + 1);

    // Helix task timeout (fallback to general task timeout for backward compatibility)
    jobConfigBuilder.setTimeoutPerTask(this.jobContext.getJobState().getPropAsLong(
        GobblinClusterConfigurationKeys.HELIX_TASK_TIMEOUT_SECONDS,
        this.jobContext.getJobState().getPropAsLong(
            ConfigurationKeys.TASK_TIMEOUT_SECONDS,
            ConfigurationKeys.DEFAULT_TASK_TIMEOUT_SECONDS)) * 1000);

    jobConfigBuilder.setFailureThreshold(workUnits.size());
    jobConfigBuilder.addTaskConfigMap(taskConfigMap).setCommand(GobblinTaskRunner.GOBBLIN_TASK_FACTORY_NAME);
    jobConfigBuilder.setNumConcurrentTasksPerInstance(ConfigUtils.getInt(jobConfig,
        GobblinClusterConfigurationKeys.HELIX_CLUSTER_TASK_CONCURRENCY,
        GobblinClusterConfigurationKeys.HELIX_CLUSTER_TASK_CONCURRENCY_DEFAULT));

    if (this.jobConfig.hasPath(GobblinClusterConfigurationKeys.HELIX_JOB_TAG_KEY)) {
      String jobTag = this.jobConfig.getString(GobblinClusterConfigurationKeys.HELIX_JOB_TAG_KEY);
      log.info("Job {} has tags associated : {}", this.jobContext.getJobId(), jobTag);
      jobConfigBuilder.setInstanceGroupTag(jobTag);
    }

    if (this.jobConfig.hasPath(GobblinClusterConfigurationKeys.HELIX_JOB_TYPE_KEY)) {
      String jobType = this.jobConfig.getString(this.jobConfig.getString(GobblinClusterConfigurationKeys.HELIX_JOB_TYPE_KEY));
      log.info("Job {} has types associated : {}", this.jobContext.getJobId(), jobType);
      jobConfigBuilder.setJobType(jobType);
    }

    if (Task.getExecutionModel(ConfigUtils.configToState(jobConfig)).equals(ExecutionModel.STREAMING)) {
      jobConfigBuilder.setRebalanceRunningTask(true);
    }

    jobConfigBuilder.setExpiry(this.jobContext.getJobState().getPropAsLong(
        GobblinClusterConfigurationKeys.HELIX_WORKFLOW_EXPIRY_TIME_SECONDS, GobblinClusterConfigurationKeys.DEFAULT_HELIX_WORKFLOW_EXPIRY_TIME_SECONDS));

    return jobConfigBuilder;
  }
1047484
public WorkUnitStream filter(Predicate<WorkUnit> predicate) {
    if (this.materializedWorkUnits == null) {
      return new BasicWorkUnitStream(this, Iterators.filter(this.workUnits, predicate), null);
    } else {
      return new BasicWorkUnitStream(this, null, Lists.newArrayList(Iterables.filter(this.materializedWorkUnits, predicate)));
    }
  }