code: private <T> Map<String, byte[]> encodeAndDistributeElements(final List<T> elements,
                                                                final List<Integer> counts,
                                                                final List<? extends Identifier> taskOrder,
                                                                final Codec<T> codec) {
      final Map<String, byte[]> taskIdToBytes = new HashMap<>();
  
      int elementsIndex = 0;
      for (int taskOrderIndex = 0; taskOrderIndex < taskOrder.size(); taskOrderIndex++) {
        final int elementCount = counts.get(taskOrderIndex);
  
        try (final ByteArrayOutputStream bstream = new ByteArrayOutputStream();
             final DataOutputStream dstream = new DataOutputStream(bstream)) {
  
          dstream.writeInt(elementCount);
          for (final T element : elements.subList(elementsIndex, elementsIndex + elementCount)) {
            final byte[] encodedElement = codec.encode(element);
            dstream.writeInt(encodedElement.length);
            dstream.write(encodedElement);
          }
          taskIdToBytes.put(taskOrder.get(taskOrderIndex).toString(), bstream.toByteArray());
  
        } catch (final IOException e) {
          throw new RuntimeException("IOException",  e);
        }
  
        elementsIndex += elementCount;
      }
  
      return taskIdToBytes;
    }
docs: Encode elements into byte arrays, and distribute them across Tasks indicated by Identifiers.
  Note that elements are distributed in the exact order specified in
  {@code elements} and not in a round-robin fashion.
  For example, (1, 2, 3, 4) uniformly distributed to (task1, task2, task3) would be
  {task1: (1, 2), task2: (3), task3: (4)}.
  
  @param elements list of data elements to encode
  @param counts list of numbers specifying how many elements each Task should receive
  @param taskOrder list of Identifiers indicating Task Ids
  @param codec class for encoding data
  @param <T> type of data
  @return byte representation of a map of identifiers to encoded data