code: @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));
          }
        }
      }
    }
docs: 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