private static void addInputPath(JobConf conf, Path path, Schema inputSchema) {

    String schemaMapping = path.toString() + ";" + toBase64(inputSchema.toString());

    String schemas = conf.get(SCHEMA_KEY);
    conf.set(SCHEMA_KEY, schemas == null ? schemaMapping : schemas + "," + schemaMapping);

    conf.setInputFormat(DelegatingInputFormat.class);
  }
@Test
  void job() throws Exception {
    JobConf job = new JobConf();
    Path inputPath1 = new Path(INPUT_DIR_1.getPath());
    Path inputPath2 = new Path(INPUT_DIR_2.getPath());
    Path outputPath = new Path(OUTPUT_DIR.getPath());

    outputPath.getFileSystem(job).delete(outputPath, true);

    writeNamesFiles(new File(inputPath1.toUri().getPath()));
    writeBalancesFiles(new File(inputPath2.toUri().getPath()));

    job.setJobName("multiple-inputs-join");
    AvroMultipleInputs.addInputPath(job, inputPath1, NamesMapImpl.class,
        ReflectData.get().getSchema(NamesRecord.class));
    AvroMultipleInputs.addInputPath(job, inputPath2, BalancesMapImpl.class,
        ReflectData.get().getSchema(BalancesRecord.class));

    Schema keySchema = ReflectData.get().getSchema(KeyRecord.class);
    Schema valueSchema = ReflectData.get().getSchema(JoinableRecord.class);
    AvroJob.setMapOutputSchema(job, Pair.getPairSchema(keySchema, valueSchema));
    AvroJob.setOutputSchema(job, ReflectData.get().getSchema(CompleteRecord.class));

    AvroJob.setReducerClass(job, ReduceImpl.class);
    job.setNumReduceTasks(1);

    FileOutputFormat.setOutputPath(job, outputPath);

    AvroJob.setReflect(job);

    JobClient.runJob(job);

    validateCompleteFile(new File(OUTPUT_DIR, "part-00000.avro"));
  }