@VisibleForTesting
  static ResolvingDecoder resolve(Decoder decoder, Schema readSchema, Schema fileSchema)
      throws IOException {
    Map<Schema, Map<Schema, ResolvingDecoder>> cache = DECODER_CACHES.get();
    Map<Schema, ResolvingDecoder> fileSchemaToResolver =
        cache.computeIfAbsent(readSchema, k -> new WeakHashMap<>());

    ResolvingDecoder resolver =
        fileSchemaToResolver.computeIfAbsent(fileSchema, schema -> newResolver(readSchema, schema));

    resolver.configure(decoder);

    return resolver;
  }
@SuppressWarnings("UnusedAssignment") // the unused assignments are necessary for this test
  @Test
  public void testDecoderCachingReadSchemaNotSameAsFileSchema() throws Exception {
    Decoder dummyDecoder = DecoderFactory.get().binaryDecoder(new byte[] {}, null);
    Schema fileSchema = avroSchema();
    Schema readSchema = avroSchema("manifest_path", "manifest_length", "partition_spec_id");
    ResolvingDecoder resolvingDecoder =
        DecoderResolver.resolve(dummyDecoder, readSchema, fileSchema);

    assertThat(DecoderResolver.DECODER_CACHES.get()).hasSize(1);
    assertThat(DecoderResolver.DECODER_CACHES.get().get(readSchema)).hasSize(1);
    checkCached(readSchema, fileSchema);

    // Equal but new one
    Schema fileSchema1 = avroSchema();
    Schema readSchema1 = avroSchema("manifest_path", "manifest_length", "partition_spec_id");
    assertThat(fileSchema1).isEqualTo(fileSchema);
    assertThat(readSchema1).isEqualTo(readSchema);
    ResolvingDecoder resolvingDecoder1 =
        DecoderResolver.resolve(dummyDecoder, readSchema1, fileSchema1);
    assertThat(resolvingDecoder1).isNotSameAs(resolvingDecoder);

    assertThat(DecoderResolver.DECODER_CACHES.get()).hasSize(2);
    assertThat(DecoderResolver.DECODER_CACHES.get().get(readSchema1)).hasSize(1);
    checkCached(readSchema1, fileSchema1);

    // New read schema
    Schema readSchema2 = avroSchema("manifest_path", "manifest_length");
    ResolvingDecoder resolvingDecoder2 =
        DecoderResolver.resolve(dummyDecoder, readSchema2, fileSchema);
    assertThat(resolvingDecoder2).isNotSameAs(resolvingDecoder);

    assertThat(DecoderResolver.DECODER_CACHES.get()).hasSize(3);
    assertThat(DecoderResolver.DECODER_CACHES.get().get(readSchema2)).hasSize(1);
    checkCached(readSchema2, fileSchema);

    checkCachedSize(3);

    readSchema = null;
    checkCachedSize(2);

    readSchema1 = null;
    checkCachedSize(1);

    readSchema2 = null;
    checkCachedSize(0);
  }