Basic support for Confluent's KafkaAvroSerializer / KafkaAvroDeSerializer

[fmck3516]

Basic support to serialize and deserialize with Confluent's proprietary 5-byte prefix.

The prefix will be added if the key or value serializer is set to io.confluent.kafka.serializers.KafkaAvroSerializer.
The prefix will be removed if the key or value deserializer is set to io.confluent.kafka.serializers.KafkaAvroDeserializer.

Configuration example:

var configuration = JSON.stringify({
    consumer: {
        keyDeserializer: "io.confluent.kafka.serializers.KafkaAvroDeserializer",
        valueDeserializer: "io.confluent.kafka.serializers.KafkaAvroDeserializer",
    },
    producer: {
        keySerializer: "io.confluent.kafka.serializers.KafkaAvroSerializer",
        valueSerializer: "io.confluent.kafka.serializers.KafkaAvroSerializer",
    },
    schemaRegistry: {
        url: "https://subdomain.us-east-2.aws.confluent.cloud",
        basicAuth: {
            credentialsSource: "USER_INFO",
            userInfo: "KEY:SECRET"
        },
    },
})

The configuration for schemaRegistry is required to use a serializer of type io.confluent.kafka.serializers.KafkaAvroSerializer.

A new produce and consume function has been added (each accepts the new configuration argument):

produceWithConfiguration(producer, messages, configuration, keySchema, valueSchema);
consumeWithConfiguration(consumer, 20, configuration, keySchema, valueSchema);

Note: I ported a sub-set of the existing functionality of the Confluent Java client. This code will be obsolete once Confluent adds Schema Registry support for their Go client. At the time of writing, the Go client does not support Schema Registry: https://docs.confluent.io/platform/current/clients/index.html

[fmck3516]

I'm not familiar with Go - I still have to figure out how to unit test the stuff.

[mostafa]

Hey @fmck3516,

Thanks for the awesome PR. 🙏 I left a few comments and I think it can be merged after the changes.

I tried to create GH Actions for testing the feature, but I couldn't find an action for running Apache Kafka and co. There's an awesome GH Action by @szkiba called xk6bundler that can bundle and distribute k6 with xk6 extensions, but mere bundling without actual testing with Kafka is not what we want though.

[mostafa]

I am wondering if you could move all the scripts (test_*.js) to the scripts directory and rename this somehow to something like test_avro_with_magic_bytes.js or something shorter.

[fmck3516]

I renamed the file to test_avro_with_schema_registry.js and yes, I think it would be good to move the tests out of the top level directory.

[mostafa]

But the file is still called test/test.js. Am I missing something? 🤔

[fmck3516]

@mostafa: Thanks for the feedback - I will update the PR in the upcoming week.

[fmck3516]

I incorporated your feedback. I also switched to a JSON-based configuration model to be in sync with the approach for the auth configuration.

What do you think about having one big configuration JSON for auth and everything else?

[mostafa]

@fmck3516 I was actually thinking about the same thing, but then thought that it's out of scope of this PR or mine. So maybe the next PR would focus on refactoring, cleanups and better configuration management. I also have the idea to refactor metric collection and error handling. Sorry for too much information! 🤣

[mostafa]

LGTM! I can rearrange test scripts (files) after merging.

[mostafa]

I love this. 👏

[mostafa]

But the file is still called test/test.js. Am I missing something? 🤔

[fmck3516]

@fmck3516 I was actually thinking about the same thing, but then thought that it's out of scope of this PR or mine.

That makes sense - one step at a time.

So maybe the next PR would focus on refactoring, cleanups and better configuration management. I also have the idea to refactor metric collection and error handling. Sorry for too much information! 🤣

One thing I would like to add to the wishlist: A suite of regression tests that is executed automatically.