@SuppressWarnings("unchecked")
  public static void addNamedOutput(Job job, String namedOutput, Class<? extends OutputFormat> outputFormatClass,
      Schema keySchema) {
    addNamedOutput(job, namedOutput, outputFormatClass, keySchema, null);
  }
@Test
  void avroGenericOutput() throws Exception {
    Job job = Job.getInstance();

    FileInputFormat.setInputPaths(job,
        new Path(getClass().getResource("/org/apache/avro/mapreduce/mapreduce-test-input.txt").toURI().toString()));
    job.setInputFormatClass(TextInputFormat.class);

    job.setMapperClass(LineCountMapper.class);
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(IntWritable.class);

    job.setReducerClass(GenericStatsReducer.class);
    AvroJob.setOutputKeySchema(job, STATS_SCHEMA);
    AvroMultipleOutputs.addNamedOutput(job, "myavro", AvroKeyOutputFormat.class, STATS_SCHEMA, null);
    AvroMultipleOutputs.addNamedOutput(job, "myavro1", AvroKeyOutputFormat.class, STATS_SCHEMA_2);
    job.setOutputFormatClass(AvroKeyOutputFormat.class);

    Path outputPath = new Path(DIR.getPath() + "/testAvroGenericOutput");
    outputPath.getFileSystem(job.getConfiguration()).delete(outputPath, true);
    FileOutputFormat.setOutputPath(job, outputPath);

    assertTrue(job.waitForCompletion(true));

    // Check that the results from the MapReduce were as expected.
    FileSystem fileSystem = FileSystem.get(job.getConfiguration());
    FileStatus[] outputFiles = fileSystem.globStatus(outputPath.suffix("/myavro-r-00000.avro"));
    assertEquals(1, outputFiles.length);

    Map<String, Integer> counts = new HashMap<>();

    try (DataFileReader<GenericData.Record> reader = new DataFileReader<>(
        new FsInput(outputFiles[0].getPath(), job.getConfiguration()), new GenericDatumReader<>(STATS_SCHEMA))) {
      for (GenericData.Record record : reader) {
        counts.put(((Utf8) record.get("name")).toString(), (Integer) record.get("count"));
      }
    }

    assertEquals(3, counts.get("apple").intValue());
    assertEquals(2, counts.get("banana").intValue());
    assertEquals(1, counts.get("carrot").intValue());

    outputFiles = fileSystem.globStatus(outputPath.suffix("/myavro1-r-00000.avro"));
    assertEquals(1, outputFiles.length);
    counts.clear();
    try (DataFileReader<GenericData.Record> reader = new DataFileReader<>(
        new FsInput(outputFiles[0].getPath(), job.getConfiguration()), new GenericDatumReader<>(STATS_SCHEMA_2))) {
      for (GenericData.Record record : reader) {
        counts.put(((Utf8) record.get("name1")).toString(), (Integer) record.get("count1"));
      }
    }

    assertEquals(3, counts.get("apple").intValue());
    assertEquals(2, counts.get("banana").intValue());
    assertEquals(1, counts.get("carrot").intValue());

    outputFiles = fileSystem.globStatus(outputPath.suffix("/testnewwrite-r-00000.avro"));
    assertEquals(1, outputFiles.length);
    counts.clear();
    try (DataFileReader<GenericData.Record> reader = new DataFileReader<>(
        new FsInput(outputFiles[0].getPath(), job.getConfiguration()), new GenericDatumReader<>(STATS_SCHEMA))) {
      for (GenericData.Record record : reader) {
        counts.put(((Utf8) record.get("name")).toString(), (Integer) record.get("count"));
      }
    }

    assertEquals(3, counts.get("apple").intValue());
    assertEquals(2, counts.get("banana").intValue());
    assertEquals(1, counts.get("carrot").intValue());

    outputFiles = fileSystem.globStatus(outputPath.suffix("/testnewwrite2-r-00000.avro"));
    assertEquals(1, outputFiles.length);
    counts.clear();
    try (DataFileReader<GenericData.Record> reader = new DataFileReader<>(
        new FsInput(outputFiles[0].getPath(), job.getConfiguration()), new GenericDatumReader<>(STATS_SCHEMA_2))) {
      for (GenericData.Record record : reader) {
        counts.put(((Utf8) record.get("name1")).toString(), (Integer) record.get("count1"));
      }
    }
    assertEquals(3, counts.get("apple").intValue());
    assertEquals(2, counts.get("banana").intValue());
    assertEquals(1, counts.get("carrot").intValue());

    outputFiles = fileSystem.globStatus(outputPath.suffix("/testwritenonschema-r-00000.avro"));
    assertEquals(1, outputFiles.length);
    counts.clear();
    try (DataFileReader<GenericData.Record> reader = new DataFileReader<>(
        new FsInput(outputFiles[0].getPath(), job.getConfiguration()), new GenericDatumReader<>(STATS_SCHEMA))) {
      for (GenericData.Record record : reader) {
        counts.put(((Utf8) record.get("name")).toString(), (Integer) record.get("count"));
      }
    }

    assertEquals(3, counts.get("apple").intValue());
    assertEquals(2, counts.get("banana").intValue());
    assertEquals(1, counts.get("carrot").intValue());
  }