ConfluentSchemaRegistryDeserializerProvider implements DeserializerProvider<T> { @Override public Coder<T> getCoder(CoderRegistry coderRegistry) { return (Coder<T>) AvroCoder.of(getAvroSchema()); } @VisibleForTesting  ConfluentSchemaRegistryDeserializerProvider(
      SerializableFunction<Void, SchemaRegistryClient> schemaRegistryClientProviderFn,
      String schemaRegistryUrl,
      String subject,
      @Nullable Integer version); static ConfluentSchemaRegistryDeserializerProvider<T> of(
      String schemaRegistryUrl, String subject); static ConfluentSchemaRegistryDeserializerProvider<T> of(
      String schemaRegistryUrl, String subject, @Nullable Integer version); @Override Deserializer<T> getDeserializer(Map<String, ?> configs, boolean isKey); @Override Coder<T> getCoder(CoderRegistry coderRegistry);  }
@Test public void testGetCoder() { String schemaRegistryUrl = "mock: String subject = "mytopic"; SchemaRegistryClient mockRegistryClient = mockSchemaRegistryClient(schemaRegistryUrl, subject); CoderRegistry coderRegistry = CoderRegistry.createDefault(); AvroCoder coderV0 = (AvroCoder) mockDeserializerProvider(schemaRegistryUrl, subject, null).getCoder(coderRegistry); assertEquals(AVRO_SCHEMA, coderV0.getSchema()); try { Integer version = mockRegistryClient.getVersion(subject, AVRO_SCHEMA_V1); AvroCoder coderV1 = (AvroCoder) 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); } }