@Override
  public int run(InputStream in, PrintStream out, PrintStream err, List<String> args) throws Exception {

    if (args.isEmpty()) {
      printHelp(out);
      return 0;
    }

    OutputStream output = out;
    if (args.size() > 1) {
      output = Util.fileOrStdout(args.get(args.size() - 1), out);
      args = args.subList(0, args.size() - 1);
    }

    DataFileWriter<GenericRecord> writer = new DataFileWriter<>(new GenericDatumWriter<>());
    Schema schema = null;
    Map<String, byte[]> metadata = new TreeMap<>();
    String inputCodec = null;

    for (String inFile : expandsInputFiles(args)) {
      InputStream input = Util.fileOrStdin(inFile, in);
      DataFileStream<GenericRecord> reader = new DataFileStream<>(input, new GenericDatumReader<>());

      if (schema == null) {
        // this is the first file - set up the writer, and store the
        // Schema & metadata we'll use.
        schema = reader.getSchema();
        for (String key : reader.getMetaKeys()) {
          if (!DataFileWriter.isReservedMeta(key)) {
            byte[] metadatum = reader.getMeta(key);
            metadata.put(key, metadatum);
            writer.setMeta(key, metadatum);
          }
        }
        inputCodec = reader.getMetaString(DataFileConstants.CODEC);
        if (inputCodec == null) {
          inputCodec = DataFileConstants.NULL_CODEC;
        }
        writer.setCodec(CodecFactory.fromString(inputCodec));
        writer.create(schema, output);
      } else {
        // check that we're appending to the same schema & metadata.
        if (!schema.equals(reader.getSchema())) {
          err.println("input files have different schemas");
          reader.close();
          return 1;
        }
        for (String key : reader.getMetaKeys()) {
          if (!DataFileWriter.isReservedMeta(key)) {
            byte[] metadatum = reader.getMeta(key);
            byte[] writersMetadatum = metadata.get(key);
            if (!Arrays.equals(metadatum, writersMetadatum)) {
              err.println("input files have different non-reserved metadata");
              reader.close();
              return 2;
            }
          }
        }
        String thisCodec = reader.getMetaString(DataFileConstants.CODEC);
        if (thisCodec == null) {
          thisCodec = DataFileConstants.NULL_CODEC;
        }
        if (!inputCodec.equals(thisCodec)) {
          err.println("input files have different codecs");
          reader.close();
          return 3;
        }
      }

      writer.appendAllFrom(reader, /* recompress */ false);
      reader.close();
    }

    writer.close();
    return 0;
  }
@Test
  void dirConcat() throws Exception {
    Map<String, String> metadata = new HashMap<>();

    for (int i = 0; i < 3; i++) {
      generateData(name.getMethodName() + "-" + i + ".avro", Type.STRING, metadata, DEFLATE);
    }

    File output = new File(OUTPUT_DIR, name.getMethodName() + ".avro");

    List<String> args = asList(INPUT_DIR.getAbsolutePath(), output.getAbsolutePath());
    int returnCode = new ConcatTool().run(System.in, System.out, System.err, args);

    assertEquals(0, returnCode);
    assertEquals(ROWS_IN_INPUT_FILES * 3, numRowsInFile(output));
  }