Merge all of the metadata output from each work-unit and publish the merged record.
@param states States from all tasks
@throws IOException If there is an error publishing the file
327323
@Override
  public void publishMetadata(Collection<? extends WorkUnitState> states)
      throws IOException {
   Set<String> partitions = new HashSet<>();

    // There should be one merged metadata file per branch; first merge all of the pieces together
    mergeMetadataAndCollectPartitionNames(states, partitions);
    partitions.removeIf(Objects::isNull);

    // Now, pick an arbitrary WorkUnitState to get config information around metadata such as
    // the desired output filename. We assume that publisher config settings
    // are the same across all workunits so it doesn't really matter which workUnit we retrieve this information
    // from.

    WorkUnitState anyState = states.iterator().next();
    for (int branchId = 0; branchId < numBranches; branchId++) {
      String mdOutputPath = getMetadataOutputPathFromState(anyState, branchId);
      String userSpecifiedPath = getUserSpecifiedOutputPathFromState(anyState, branchId);

      if (partitions.isEmpty() || userSpecifiedPath != null) {
        publishMetadata(getMergedMetadataForPartitionAndBranch(null, branchId),
            branchId,
            getMetadataOutputFileForBranch(anyState, branchId));
      } else {
        String metadataFilename = getMetadataFileNameForBranch(anyState, branchId);
        if (mdOutputPath == null || metadataFilename == null) {
          LOG.info("Metadata filename not set for branch " + String.valueOf(branchId) + ": not publishing metadata.");
          continue;
        }

        for (String partition : partitions) {
          publishMetadata(getMergedMetadataForPartitionAndBranch(partition, branchId),
              branchId,
              new Path(new Path(mdOutputPath, partition), metadataFilename));
        }
      }
    }
  }
1010117
public void addMasterState(MasterState state) {
		checkNotNull(state);

		synchronized (lock) {
			if (!discarded) {
				masterState.add(state);
			}
		}
	}
667942
public function push($type, $state = []) {
        if (isset($this->_defaults[$type])) {
            $state = array_merge($this->_defaults[$type], $state);
        }
        array_push($this->_states, [$type, $state]);
    }
996378
public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeBoolean(isMaster);
        dataOutput.writeUTF(serverState);
        dataOutput.writeInt(totalUpdates);
        dataOutput.writeInt(streamId);

    }
847892
@Deprecated
	public void setViewStates(@NonNull final SparseArray<ViewState> states) {
		mViewStates.clear();
		for (int i = 0; i < states.size(); i++) {
			final int key = states.keyAt(i);
			final ViewState value = states.get(key);
			mViewStates.put(key, value);
		}
	}
411274
public static void mergeFiles(File[] files, File destinationFile, String[] filterRegex) throws IOException {
        if (filterRegex.length < files.length) {
            filterRegex = ArrayUtils.concatArrays(filterRegex, new String[files.length - filterRegex.length]);
        }
        createNewFile(destinationFile);
        int i = 0;
        for (File file : files) {
            mergeFiles(file, destinationFile, filterRegex[i++]);
        }
    }
1299326
@VisibleForTesting
	void onPartitionStateUpdate(
			IntermediateDataSetID intermediateDataSetId,
			ResultPartitionID resultPartitionId,
			ExecutionState producerState) throws IOException, InterruptedException {

		if (executionState == ExecutionState.RUNNING) {
			final SingleInputGate inputGate = inputGatesById.get(intermediateDataSetId);

			if (inputGate != null) {
				if (producerState == ExecutionState.SCHEDULED
					|| producerState == ExecutionState.DEPLOYING
					|| producerState == ExecutionState.RUNNING
					|| producerState == ExecutionState.FINISHED) {

					// Retrigger the partition request
					inputGate.retriggerPartitionRequest(resultPartitionId.getPartitionId());

				} else if (producerState == ExecutionState.CANCELING
					|| producerState == ExecutionState.CANCELED
					|| producerState == ExecutionState.FAILED) {

					// The producing execution has been canceled or failed. We
					// don't need to re-trigger the request since it cannot
					// succeed.
					if (LOG.isDebugEnabled()) {
						LOG.debug("Cancelling task {} after the producer of partition {} with attempt ID {} has entered state {}.",
							taskNameWithSubtask,
							resultPartitionId.getPartitionId(),
							resultPartitionId.getProducerId(),
							producerState);
					}

					cancelExecution();
				} else {
					// Any other execution state is unexpected. Currently, only
					// state CREATED is left out of the checked states. If we
					// see a producer in this state, something went wrong with
					// scheduling in topological order.
					String msg = String.format("Producer with attempt ID %s of partition %s in unexpected state %s.",
						resultPartitionId.getProducerId(),
						resultPartitionId.getPartitionId(),
						producerState);

					failExternally(new IllegalStateException(msg));
				}
			} else {
				failExternally(new IllegalStateException("Received partition producer state for " +
						"unknown input gate " + intermediateDataSetId + "."));
			}
		} else {
			LOG.debug("Task {} ignored a partition producer state notification, because it's not running.", taskNameWithSubtask);
		}
	}
774990
void saveState(Bundle outState) {
        outState.putString(STATE_CURRENT_PATH, currentPath);
        outState.putSerializable(STATE_FOLDERS, folders);
        outState.putSerializable(STATE_NEXT_TOKENS, nextTokens);
    }
442224
public static void cleanStagingData(List<? extends State> states, Logger logger) throws IOException {
    for (State state : states) {
      JobLauncherUtils.cleanTaskStagingData(state, logger);
    }
  }
603630
@Override
  public void write(Fields fields) throws IOException {
    delegateFieldsConsumer.write(fields);
    write(state.fieldInfos, fields);
  }
1294447
@Override
	protected List<V> mergeState(List<V> a, List<V> b) {
		a.addAll(b);
		return a;
	}
1074988
public void deleteStates(Collection<MutableState> statesToDelete) {
    if (statesToDelete.isEmpty()) {
      return;
    }
    for (MutableState state : statesToDelete) {
      deleteState(state);
    }
    remapStateIds();
  }
306177
private void updateState(final TaskStatus.State state) {
    statusUpdater.setState(state);
    // Commit and push a new status
    try {
      statusUpdater.update();
    } catch (InterruptedException e) {
      // TODO: propagate interrupt instead?
      Thread.currentThread().interrupt();
    }
  }
107511
public void writeState(DataOutputStream stream) throws IOException
        {
        int len = mt.length;
        for(int x=0;x<len;x++) stream.writeInt(mt[x]);
        
        len = mag01.length;
        for(int x=0;x<len;x++) stream.writeInt(mag01[x]);
        
        stream.writeInt(mti);
        stream.writeDouble(__nextNextGaussian);
        stream.writeBoolean(__haveNextNextGaussian);
        }
897394
public void registerAllPossibleTransition(Automaton[] states) {
        for (int i = 0; i < states.length; i++) {
            for (int j = i; j < states.length; j++) {
                registerTransition(states[i], states[j]);
                registerTransition(states[j], states[i]);
            }
        }
    }
230643
def publish_state(self, state):
        
        message = json.dumps({'state': {'reported': state}})
        self.client.publish(self.topic, message)
        self._state = state
496101
protected void init(SourceState state) {
    retriever.init(state);

    try {
      initFileSystemHelper(state);
    } catch (FileBasedHelperException e) {
      Throwables.propagate(e);
    }

    AvroFsHelper fsHelper = (AvroFsHelper) this.fsHelper;
    this.fs = fsHelper.getFileSystem();

    this.sourceState = state;

    this.lowWaterMark =
        getLowWaterMark(state.getPreviousWorkUnitStates(), state.getProp(DATE_PARTITIONED_SOURCE_MIN_WATERMARK_VALUE,
            String.valueOf(DEFAULT_DATE_PARTITIONED_SOURCE_MIN_WATERMARK_VALUE)));

    this.maxFilesPerJob = state.getPropAsInt(DATE_PARTITIONED_SOURCE_MAX_FILES_PER_JOB,
        DEFAULT_DATE_PARTITIONED_SOURCE_MAX_FILES_PER_JOB);

    this.maxWorkUnitsPerJob = state.getPropAsInt(DATE_PARTITIONED_SOURCE_MAX_WORKUNITS_PER_JOB,
        DEFAULT_DATE_PARTITIONED_SOURCE_MAX_WORKUNITS_PER_JOB);

    this.tableType = TableType.valueOf(state.getProp(ConfigurationKeys.EXTRACT_TABLE_TYPE_KEY).toUpperCase());

    this.fileCount = 0;

    this.sourceDir = new Path(state.getProp(ConfigurationKeys.SOURCE_FILEBASED_DATA_DIRECTORY));
 }