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

    FileInputFormat.setInputPaths(job,
        new Path(getClass().getResource("/org/apache/avro/mapreduce/mapreduce-test-input.avro").toURI().toString()));
    job.setInputFormatClass(AvroKeyInputFormat.class);
    AvroJob.setInputKeySchema(job, TextStats.SCHEMA$);
    AvroMultipleOutputs.addNamedOutput(job, "myavro3", AvroKeyOutputFormat.class, TextStats.SCHEMA$, null);

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

    job.setReducerClass(SpecificStatsReducer.class);
    AvroJob.setOutputKeySchema(job, TextStats.SCHEMA$);

    job.setOutputFormatClass(AvroKeyOutputFormat.class);
    Path outputPath = new Path(DIR.getPath() + "/testAvroInput");
    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("/myavro3-*"));
    assertEquals(1, outputFiles.length);
    Map<String, Integer> counts = new HashMap<>();
    try (DataFileReader<TextStats> reader = new DataFileReader<>(
        new FsInput(outputFiles[0].getPath(), job.getConfiguration()), new SpecificDatumReader<>())) {
      for (TextStats record : reader) {
        counts.put(record.getName().toString(), record.getCount());
      }
    }

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