[FLINK-33058][formats] Add encoding option to Avro format

[dalelane]

What is the purpose of the change

Initially proposed in https://issues.apache.org/jira/browse/FLINK-33058

Avro supports two serialization encoding methods: binary and JSON (cf. Avro docs)

flink-avro currently has a hard-coded assumption that Avro data is binary-encoded (and cannot process Avro data that has been JSON-encoded).

This pull request introduces a new optional format option to flink-avro: avro.encoding
It supports two options: 'binary' and 'json'.
It unset, it will default to 'binary' to maintain compatibility/consistency with current behaviour.

Brief change log

Flink uses Avro Decoder and Encoder classes for deserializing/serializing Avro data.

However it was hard-coding the use of factory classes to only use the binary-encoding implementations of these abstract classes. (DecoderFactory.get().binaryDecoder and EncoderFactory.get().directBinaryEncoder)

In this pull request, I'm using the value of the new avro.encoding option to create the JSON Decoder/Encoder classes where appropriate.

Verifying this change

This change modified existing tests by re-running all of the tests that perform Avro serialization/deserialization to repeat the test using both binary and avro encoding.

This verifies that the existing binary behaviour is unaffected by the new option, as well as the new JSON support.

I've also manually verified the new support using Flink SQL such as:

CREATE TABLE JSONAVRO
(
    ... my columns ...
)
WITH (
    'connector' = 'kafka',
    'topic' = 'MY.TOPIC',
    'properties.bootstrap.servers' = 'localhost:9092',
    'properties.group.id' = 'my-group',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'avro',
    'avro.encoding' = 'json'
);

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): yes
• The serializers: no
• The runtime per-record code paths (performance sensitive): yes
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? update to the Avro format page to explain the new option

[flinkbot]

CI report:

• 5b1c071 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[dalelane]

Apologies for the flurry of follow-on commits - this is my first contribution to Flink so I'd missed the checkstyle and spotless rules when testing locally.

I think it's ready to review now, but I'm sure there are still other things I've unwittingly missed! Please let me know if there is anything else that I should do to get this PR into an acceptable state.

[afedulov]

Hi @dalelane, thanks for your contribution. We need to always consider what happens when the user tries to compile his old user code against the new version of Flink. As such, merging your changes would remove public APIs and is not backwards compatible. Public methods can only be removed though a deprecation process.

[RyanSkraba]

Hello! Overall, this does exactly what it says -- it looks good! Thanks for checking the public API, but I think I found one more in AvroSerializationSchema.

Other than that, there's a couple of minor suggestions that you can take or leave, especially the @ParameterizedTest. This is something that I largely prefer, but it makes little difference when all the tests are passing!

I mentioned in the Jira: using JSON-encoded Avro is really noot a best practice, but it does provide human-readable messages with schema-enforced structure... I guess we could put a warning someplace in the code, but if that's what customers are looking for, it's hard to argue!

That being said, @afedulov, do you think it's worthwhile bringing up the new feature on the mailing list to discuss?

[dalelane]

Thanks for the reviews - much appreciated 👍

[afedulov]

That being said, @afedulov, do you think it's worthwhile bringing up the new feature on the mailing list to discuss?

@RyanSkraba This was my initial thought, yes. Ideally we do not want to introduce functionality for very niche use cases, but this one makes sense to me, especially for building demos etc. Although this change, in my opinion, does not deserve a FLIP, I think it still makes sense to do a quick vote in the dev mailing list. The idea would be to prepend the topic with [VOTE], briefly describe the proposal, why it is useful and the downsides of it not being the best practice (Ryan's concerns). If no one comments - this is a silent yes.

[RyanSkraba]

LGTM -- thanks!

[RyanSkraba]

I took the liberty of bringing it up in the mailing list -- I thought that was fair!

[JingGe]

LGTM. Thanks for driving it!

[dalelane]

@afedulov Is there anything else that you think is needed here?

[afedulov]

LGTM

[davidradl]

@afedulov @JingGe @RyanSkraba It looks like this is approved by all and ready to go for some time now, please could one of you merge this - so this is not forgotten. Many thanks.

[afedulov]

From us three only Jing has the permissions to merge.

[JingGe]

@dalelane would you please rebase and squash the commits? Once the CI passed, I will merge it. Thanks!

[dalelane]

@JingGe Thanks very much - I've done the squash and rebase now