@Override
  public Coder<T> getCoder(CoderRegistry coderRegistry) {
    return (Coder<T>) AvroCoder.of(getAvroSchema());
  }
@Test
  public void testGetCoder() {
    String schemaRegistryUrl = "mock://my-scope-name";
    String subject = "mytopic";
    SchemaRegistryClient mockRegistryClient = mockSchemaRegistryClient(schemaRegistryUrl, subject);
    CoderRegistry coderRegistry = CoderRegistry.createDefault();

    AvroCoder<Object> coderV0 =
        (AvroCoder<Object>)
            mockDeserializerProvider(schemaRegistryUrl, subject, null).getCoder(coderRegistry);
    assertEquals(AVRO_SCHEMA, coderV0.getSchema());

    try {
      Integer version = mockRegistryClient.getVersion(subject, AVRO_SCHEMA_V1);
      AvroCoder<Object> coderV1 =
          (AvroCoder<Object>)
              mockDeserializerProvider(schemaRegistryUrl, subject, version).getCoder(coderRegistry);
      assertEquals(AVRO_SCHEMA_V1, coderV1.getSchema());
    } catch (IOException | RestClientException e) {
      throw new RuntimeException("Unable to register schema for subject: " + subject, e);
    }
  }