@SuppressWarnings("squid:S135")
        @Override
        public void run() {
          LOGGER.debug(
              "Database {} memtable flushing to file {} starts to encoding data.",
              storageGroup,
              writer.getFile().getName());
          while (true) {

            Object task;
            try {
              task = encodingTaskQueue.take();
            } catch (InterruptedException e1) {
              LOGGER.error("Take task into ioTaskQueue Interrupted");
              Thread.currentThread().interrupt();
              break;
            }
            if (task instanceof StartFlushGroupIOTask || task instanceof EndChunkGroupIoTask) {
              try {
                ioTaskQueue.put(task);
              } catch (
                  @SuppressWarnings("squid:S2142")
                  InterruptedException e) {
                LOGGER.error(
                    "Database {} memtable flushing to file {}, encoding task is interrupted.",
                    storageGroup,
                    writer.getFile().getName(),
                    e);
                // generally it is because the thread pool is shutdown so the task should be aborted
                break;
              }
            } else if (task instanceof TaskEnd) {
              break;
            } else {
              long starTime = System.currentTimeMillis();
              IWritableMemChunk writableMemChunk = (IWritableMemChunk) task;
              if (writableMemChunk instanceof AlignedWritableMemChunk && times == null) {
                times = new long[MAX_NUMBER_OF_POINTS_IN_PAGE];
              }
              writableMemChunk.encode(ioTaskQueue, encodeInfo, times);
              long subTaskTime = System.currentTimeMillis() - starTime;
              WRITING_METRICS.recordFlushSubTaskCost(WritingMetrics.ENCODING_TASK, subTaskTime);
              memSerializeTime += subTaskTime;
            }
          }
          try {
            ioTaskQueue.put(new TaskEnd());
          } catch (InterruptedException e) {
            LOGGER.error("Put task into ioTaskQueue Interrupted");
            Thread.currentThread().interrupt();
          }

          DataRegion.getNonSystemDatabaseName(storageGroup)
              .ifPresent(
                  databaseName ->
                      recordFlushPointsMetricInternal(
                          memTable.getTotalPointsNum(), databaseName, dataRegionId));
          WRITING_METRICS.recordFlushCost(WritingMetrics.FLUSH_STAGE_ENCODING, memSerializeTime);
        }
encoding task (second task of pipeline)
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java