@Override
    public TaskDataWriter<T> getDataWriter(DataWriterKey writerKey) throws IOException {
      TaskDataWriter<T> writer;
      writer = dataWriterMap.get(writerKey);
      if (writer != null && shouldRollToNewFile(writer)) {
        writer.close();
        completedFiles.add(writer.toDataFile());
        dataWriterMap.remove(writerKey);
      }

      if (!dataWriterMap.containsKey(writerKey)) {
        DataWriterKey copiedWriterKey = writerKey.copy();
        writer = newWriter(copiedWriterKey);
        dataWriterMap.put(copiedWriterKey, writer);
      } else {
        writer = dataWriterMap.get(writerKey);
      }
      return writer;
    }
a fan-out writer holder which will keep an opened writer for all write key. This holder does
not require records have been sorted, but will keep open files as many as write keys.
amoro-format-iceberg/src/main/java/org/apache/amoro/io/writer/BaseTaskWriter.java