MartijnVisser commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r3960932383
##########
flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/ConfluentSchemaRegistryCoder.java:
##########
@@ -81,13 +100,29 @@ public Schema readSchema(InputStream in) throws
IOException {
@Override
public void writeSchema(Schema schema, OutputStream out) throws
IOException {
- try {
- int registeredId = schemaRegistryClient.register(subject, schema);
- out.write(CONFLUENT_MAGIC_BYTE);
- byte[] schemaIdBytes =
ByteBuffer.allocate(4).putInt(registeredId).array();
- out.write(schemaIdBytes);
- } catch (RestClientException e) {
- throw new IOException("Could not register schema in registry", e);
+ int registeredId;
+ if (registerSchema()) {
+ try {
+ registeredId = schemaRegistryClient.register(subject, schema);
+ } catch (RestClientException e) {
+ throw new IOException("Could not register schema in registry",
e);
+ }
+ } else {
+ try {
+ registeredId = schemaRegistryClient.getId(subject, schema);
+ } catch (RestClientException e) {
+ throw new IOException("Could not retrieve schema in registry",
e);
Review Comment:
Yes, `getId` throws a RestClientException with 40401 and that's wrapped in
an IOException. I kept IOException since that's what the SchemaCoder interface
declares, but the message now includes the schema name and subject and states
that the schema has to be registered beforehand when the flag is false.
##########
flink-formats/flink-avro-confluent-registry/src/test/java/org/apache/flink/formats/avro/registry/confluent/RegistryAvroFormatFactoryTest.java:
##########
@@ -229,6 +229,74 @@ public void
testSerializationSchemaWithInvalidOptionalSchema() {
null, SCHEMA.toPhysicalRowDataType()));
}
+ @Test
Review Comment:
Turned into one parameterized test over true/false that checks both source
and sink. The variant that passes the flag through the properties map stays a
separate test.
##########
flink-end-to-end-tests/flink-confluent-schema-registry/src/main/resources/avro/input-record.avsc:
##########
@@ -15,13 +15,14 @@
* limitations under the License.
*/
- {"namespace": "example.avro",
- "type": "record",
- "name": "User",
- "fields": [
- {"name": "name", "type": "string", "default": ""},
- {"name": "favoriteNumber", "type": "string", "default": ""},
- {"name": "favoriteColor", "type": "string", "default": ""},
- {"name": "eventType","type": {"name": "EventType","type": "enum",
"symbols": ["meeting"] }}
- ]
-}
+{
+ "namespace": "org.apache.flink.avro.generated",
+ "type": "record",
+ "name": "record",
+ "fields": [
+ {"name": "name", "type": ["null", "string"], "default": null},
Review Comment:
The record name matters as well: with the flag set to false the schema Flink
infers has to match what's in the registry byte for byte, which is why the docs
recommend setting `avro-confluent.schema` explicitly.
##########
docs/content/docs/connectors/table/formats/avro-confluent.md:
##########
@@ -287,6 +287,14 @@ Format Options
<td>String</td>
<td>The URL of the Confluent Schema Registry to fetch/register
schemas.</td>
</tr>
+ <tr>
+ <td><h5>auto.register.schemas</h5></td>
+ <td>optional</td>
+ <td>yes</td>
Review Comment:
Added the row to the Chinese docs as well.
##########
flink-python/pom.xml:
##########
@@ -289,7 +289,7 @@ under the License.
<!-- Indirectly accessed in pyflink_gateway_server -->
<groupId>org.apache.flink</groupId>
<artifactId>flink-sql-connector-kafka</artifactId>
- <version>3.0.0-1.17</version>
+ <version>4.0.0-2.0</version>
Review Comment:
Dropped from this PR, master is already on 5.0.0-2.2.
##########
flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/ConfluentSchemaRegistryCoder.java:
##########
@@ -46,9 +50,24 @@ public class ConfluentSchemaRegistryCoder implements
SchemaCoder {
* @param schemaRegistryClient client to connect schema registry
* @param subject subject of schema registry to produce
*/
- public ConfluentSchemaRegistryCoder(String subject, SchemaRegistryClient
schemaRegistryClient) {
+ public ConfluentSchemaRegistryCoder(
+ String subject,
+ SchemaRegistryClient schemaRegistryClient,
+ @Nullable Map<String, ?> registryConfigs) {
this.schemaRegistryClient = schemaRegistryClient;
this.subject = subject;
+ this.registryConfigs = registryConfigs;
+ }
+
+ /**
+ * Creates {@link SchemaCoder} that uses provided {@link
SchemaRegistryClient} to connect to
+ * schema registry.
+ *
+ * @param schemaRegistryClient client to connect schema registry
Review Comment:
Done.
##########
flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/AvroConfluentFormatOptions.java:
##########
@@ -65,6 +65,16 @@ public class AvroConfluentFormatOptions {
// Commonly used options maintained by Flink for convenience
//
--------------------------------------------------------------------------------------------
+ public static final ConfigOption<Boolean> AUTO_REGISTER_SCHEMAS =
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
