Schema Registery Issue with Kafka Streams TopologyTestDriver with Avro record

user_12491669 0
Viewed 1256

I am trying to test the kafka streams using the TopologyTestDriver. I am sharing the code snippet and the error I am facing.

public class ToplogyTest extends AvroSourceJsonTopologyTestSupport {
    private static final String MOCK_SCHEMA_REGISTRY_URL = "mock://test:8081";
    private TopologyTestDriver testDriver;
    private TestInputTopic<String, GenericRecord> inputTopic;
    private MockSchemaRegistryClient schemaRegistryClient;
}

  private final Properties props;

public ToplogyTest () {
    super();
     props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streamsTest");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:1234");
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, GenericAvroSerde.class);
    props.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, MOCK_SCHEMA_REGISTRY_URL); 
    schemaRegistryClient = new MockSchemaRegistryClient();
}

 @BeforeEach
   public void setup() throws Exception {
// Created the topology
 // Create test driver
        testDriver = new TopologyTestDriver(topology, props);
        // Create Serdes used for test record keys and values
        Serde<String> stringSerde = Serdes.String();
        Serde<GenericRecord> avroSerde = new GenericAvroSerde();
final Map<String,String> avroSerdeConfig = new HashMap<>();
        avroSerdeConfig.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, MOCK_SCHEMA_REGISTRY_URL);
        avroSerde.configure(avroSerdeConfig, false);
        usersTopic = testDriver.createInputTopic(
                "input-topic",
                stringSerde.serializer(),
                avroSerde.serializer());
}
  @Test
  public void Test(){
 usersTopic.pipeInput("null",record);
}

Error org.apache.kafka.common.errors.SerializationException: Error serializing Avro message Suppressed: java.lang.IllegalArgumentException: Please always: Initialize the test driver at the end of setup before each test using provided method; Close the test driver after each test using provided method. Caused by: java.net.MalformedURLException: unknown protocol: mock at java.net.URL.(URL.java:618) at java.net.URL.(URL.java:508) at java.net.URL.(URL.java:457) at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java: 152)

What i understood here is I am not able to register the mock schema registry. Is there anyone faced similar issue ?

apache-kafka-streams apache-kafka java
2 Answers

I haven't tried with GenericAvroSerde, but with SpecificAvroSerde it has a constructor takes a param of the SchemaRegistryClient.

When you create GenericAvroSerde, pass in your instance of mockSchemaRegistryClient so that it uses your mock, to force it to not use the SchemaRegistryClient that it will create itself (with the no arg constructor).

Also, remove the property for default.value.serde=GenericAvroSerde - if you configure this in your test like you normally would in your code under test, you configure an instance of GenericAvroSerde without the mock again, and at runtime it still attempts to connect to the schema.registry.url

I don't know if this is how MockSchemaRegistryClient is intended to be used, but this approach worked for me when used with SpecificAvroSerde

user_4062900

According to the error, mock:// is not a valid URI scheme. For your test, it'll need to be http:// even if it's not used

user_23086830
Related