[FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry

[MartijnVisser]

What is the purpose of the change

This PR is based on #25410 and aims to complete the necessary tasks. It introduces auto.register.schemas as a table option. Compared to the linked PR, it includes unit tests, a new IT case and updated documentation

Brief change log

• Introduces new table option auto.register.schemas
• Adds unit tests
• Adds a new AvroConfluentITCase
• Removes previous (currently disabled) bash-based tests
• It also bumps certain dependencies

Verifying this change

This change added tests and can be verified as follows:

• Run AvroConfluentITCase that writes and reads from/to Kafka using avro-confluent, with the table option set to true (default) to show that Flink can register the schema and false where it relies on schema registration outside of Flink

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

• Dependencies (does it add or upgrade a dependency): yes
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• 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? docs / JavaDocs

[flinkbot]

CI report:

• df1d162 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[MartijnVisser]

@flinkbot run azure

[MartijnVisser]

@flinkbot run azure

[fapaul]

Thanks for working on this feature. It's a great quality of life improvement.

[raphaelauv]

hi , will this PR correct this problem ? -> https://stackoverflow.com/questions/79738462/flink-confluentregistryavroserializationschema-not-respecting-registryconfigs ? thanks

[raphaelauv]

can't we rely on the schema-registry ser/deser that already support these options like (
AUTO_REGISTER_SCHEMAS , USE_SCHEMA_ID , NORMALIZE_SCHEMAS , AVRO_REMOVE_JAVA_PROPS_CONFIG ) than adding it here ?

[MartijnVisser]

No. Flink doesn't use Confluent's KafkaAvroSerializer, ConfluentSchemaRegistryCoder talks to the SchemaRegistryClient directly and writes the magic byte and id itself. The registry configs only ever reached the client, so serializer options like auto.register.schemas were silently ignored. This PR implements that one under the same key Confluent uses, so it also works when passed via registryConfigs in the DataStream API. use.schema.id and normalize.schemas would need the same treatment, that's a follow-up.

[MartijnVisser]

@raphaelauv Yes, that's the same problem. registryConfigs only configures the client, it doesn't affect how Flink writes records. With this PR auto.register.schemas in that map is honored.