@VisibleForTesting
  List<Event> createEventList(boolean sendSerializedEvents, InputSplitInfoMem inputSplitInfo)
      throws IOException {

    List<Event> events = Lists.newArrayListWithCapacity(inputSplitInfo.getNumTasks() + 1);

    InputConfigureVertexTasksEvent configureVertexEvent =
        InputConfigureVertexTasksEvent.create(inputSplitInfo.getNumTasks(),
        VertexLocationHint.create(inputSplitInfo.getTaskLocationHints()),
        InputSpecUpdate.getDefaultSinglePhysicalInputSpecUpdate());
    events.add(configureVertexEvent);

    if (sendSerializedEvents) {
      int count = 0;
      long inMemoryPayloadSize = 0;
      long serializedPayloadSize = 0;

      int payloadSerializationThresholdBytes =
          HiveConf.getIntVar(jobConf, HiveConf.ConfVars.HIVE_TEZ_INPUT_FS_SERIALIZATION_THRESHOLD);

      List<MRSplitProto> splits = inputSplitInfo.getSplitsProto().getSplitsList();
      LOG.debug("Start creating events for {} splits", splits.size());
      long startTime = Time.monotonicNow();

      try (SplitSerializer splitSerializer = getSplitSerializer()) {
        for (MRSplitProto mrSplit : splits) {
          ByteBuffer payloadBuffer = mrSplit.toByteString().asReadOnlyByteBuffer();
          int payloadSize = payloadBuffer.limit();
          boolean shouldSerializeEventToFile =
              payloadSerializationThresholdBytes != -1 && inMemoryPayloadSize >= payloadSerializationThresholdBytes;
          LOG.debug("Split #{}, byteBuffer size: {} bytes", count, payloadSize);

          InputDataInformationEvent diEvent = null;

          if (shouldSerializeEventToFile) {
            LOG.debug("Serialize to path: {}, current in-memory payload: {} bytes >= threshold: {} bytes",
                splitSerializer.getSerializedFilePath(count), inMemoryPayloadSize, payloadSerializationThresholdBytes);
            serializedPayloadSize += payloadSize;
            diEvent = splitSerializer.write(count, mrSplit);
          } else {
            inMemoryPayloadSize += payloadSize;
            diEvent = InputDataInformationEvent.createWithSerializedPayload(count, payloadBuffer);
          }
          events.add(diEvent);
          count += 1;
        }
      }

      // this is useful for making decisions regarding split serialization threshold
      long elapsed = Time.monotonicNow() - startTime;
      LOG.info(
          "Finished creating events ({} splits) in {} ms, size of payloads: in memory: {} bytes, serialized to fs: {} bytes",
          splits.size(), elapsed, inMemoryPayloadSize, serializedPayloadSize);
    } else {
      int count = 0;
      for (org.apache.hadoop.mapred.InputSplit split : inputSplitInfo.getOldFormatSplits()) {
        InputDataInformationEvent diEvent = InputDataInformationEvent.createWithObjectPayload(
            count++, split);
        events.add(diEvent);
      }
    }
    return events;
  }
SplitSerializer is a helper class for taking care of serializing splits to the tez scratch dir
when a size criteria defined by "hive.tez.input.fs.serialization.threshold" is met.
It utilizes an ExecutorService for parallel writes to prevent a single split write operation
becoming the bottleneck (as write() is called from a loop currently).
ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HiveSplitGenerator.java