Creating test data from Confluent Control Center JSON representation

user_1313335 0
Viewed 51

I'm trying to write some unit tests for Kafka Streams and have a number of quite complex schemas that I need to incorporate into my tests.

Instead of just creating objects from scratch each time, I would ideally like to instantiate using some real data and perform tests on that. We use Confluent with records in Avro format, and can extract both schema and a text JSON-like representation from the Control Center application. The JSON is valid JSON, but it's not really in the form that you'd write it in if you were just writing JSON representations of the data, so I assume it's some representation of the underlying AVRO in text form.

I've already used the schema to create a Java SpecificRecord class (price_assessment) and would like to use the JSON string copied from the Control Center message to populate a new instance of that class to feed into to my unit test InputTopic.

The code I've tried so far is

            var testAvroString = "{JSON copied from Control Center topic}";
            Schema schema = price_assessment.getClassSchema();
            DecoderFactory decoderFactory = new DecoderFactory();
            Decoder decoder = null;
            try {
                DatumReader<price_assessment> reader = new SpecificDatumReader<price_assessment>();
                decoder = decoderFactory.get().jsonDecoder(schema, testAvroString);
                return reader.read(null, decoder);
            } catch (Exception e)
            {
                return null;
            }

which is adapted from another SO answer that was using GenericRecords. When I try running this though I get the exception Cannot invoke "org.apache.avro.Schema.equals(Object)" because "writer" is null on the reader.read(...) step.

I'm not massively familiar with streams testing or Java and I'm not sure what exactly I've done wrong. Written in Java 17, streams 3.1.0, though flexible with version

confluent-platform apache-kafka-streams avro java
1 Answers

The solution that I've managed to come up with is the following, which seems to work:

 private static <T> T avroStringToInstance(Schema classSchema, String testAvroString) {

    DecoderFactory decoderFactory = new DecoderFactory();
    GenericRecord genericRecord = null;
    try {
        Decoder decoder = decoderFactory.jsonDecoder(classSchema, testAvroString);
        DatumReader<GenericData.Record> reader =
                new GenericDatumReader<>(classSchema);
        genericRecord = reader.read(null, decoder);
    } catch (Exception e)
    {
        return null;
    }

    var specific = (T) SpecificData.get().deepCopy(genericRecord.getSchema(), genericRecord);
    return specific;
}
user_13133350
Related