You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "MartijnVisser (via GitHub)" <gi...@apache.org> on 2025/06/10 16:55:27 UTC
[PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
MartijnVisser opened a new pull request, #26662:
URL: https://github.com/apache/flink/pull/26662
## What is the purpose of the change
This PR is based on https://github.com/apache/flink/pull/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
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on PR #26662:
URL: https://github.com/apache/flink/pull/26662#issuecomment-2960178773
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "raphaelauv (via GitHub)" <gi...@apache.org>.
raphaelauv commented on PR #26662:
URL: https://github.com/apache/flink/pull/26662#issuecomment-3195862138
hi , is this PR will fix this -> https://stackoverflow.com/questions/79738462/flink-confluentregistryavroserializationschema-not-respecting-registryconfigs ? thanks
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2174563238
##########
docs/content.zh/docs/connectors/table/formats/avro-confluent.md:
##########
@@ -280,6 +280,13 @@ Format 参数
<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 style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether to automatically register schemas with the Confluent Schema Registry if they don't exist. When set to <code>false</code>, schemas must be manually registered in the Schema Registry before being used. When set to <code>true</code>, schemas will be automatically registered during serialization if they don't already exist. The default value is <code>true</code>.</td>
Review Comment:
@fapaul I guess saying serialization implies that we creating an outbound flow with the schema id- e.g. writing to a sink. We could document this more explicitly.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #26662:
URL: https://github.com/apache/flink/pull/26662#issuecomment-2960006630
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "64101b15b9806691220cec07de4da5a047aaf115",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "64101b15b9806691220cec07de4da5a047aaf115",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 64101b15b9806691220cec07de4da5a047aaf115 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2159125650
##########
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:
Disabling auto registration only means that Flink won't try to register the schema in Schema Registry during every run. However, it still means that the schema that has been registered in Schema Registry by external service, is either exactly how Flink would have registered it (so with that specific namespace), or how the user would have provided it via the `avro-confluent.schema` table property.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2159144457
##########
flink-end-to-end-tests/flink-sql-client-test/pom.xml:
##########
@@ -61,14 +61,21 @@ under the License.
<!-- Used by maven-dependency-plugin -->
<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>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-sql-avro-confluent-registry</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
- <version>3.2.3</version>
+ <version>3.9.0</version>
Review Comment:
I don't think we should, given that the Flink Kafka connector v4.0 itself refers on Kafka Client 3.9.0 https://github.com/apache/flink-connector-kafka/blob/v4.0.0/pom.xml#L53
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2141950549
##########
docs/content.zh/docs/connectors/table/formats/avro-confluent.md:
##########
@@ -280,6 +280,13 @@ Format 参数
<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>
Review Comment:
Yep good call-out, I've added it
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2159110624
##########
flink-end-to-end-tests/flink-confluent-schema-registry/pom.xml:
##########
@@ -43,101 +43,160 @@ under the License.
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-streaming-java</artifactId>
+ <artifactId>flink-table-common</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
- <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka -->
+
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-kafka</artifactId>
- <version>3.0.0-1.17</version>
+ <artifactId>flink-table-api-java-bridge</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
</dependency>
- <!-- Make sure that Shaded Guava matches the one used in the flink-connector-kafka,
- or remove when FLINK-32462 is resolved -->
+
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-shaded-guava</artifactId>
- <version>30.1.1-jre-16.1</version>
+ <artifactId>flink-end-to-end-tests-common</artifactId>
+ <version>${project.version}</version>
</dependency>
+
+ <!-- This enables the WebUI during tests. -->
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-avro</artifactId>
+ <artifactId>flink-runtime-web</artifactId>
<version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- The following dependencies are for connector/format sql-jars that
+ we copy using the maven-dependency-plugin. When extending the test
+ to cover more connectors/formats, add a dependency here and an entry
+ to the dependency-plugin configuration below.
+ This ensures that all modules we actually need (as defined by the
+ dependency-plugin configuration) are built before this module. -->
+ <dependency>
+ <!-- Used by maven-dependency-plugin -->
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-sql-connector-kafka</artifactId>
+ <version>4.0.0-2.0</version>
+ <scope>provided</scope>
</dependency>
+
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-avro-confluent-registry</artifactId>
+ <artifactId>flink-sql-avro-confluent-registry</artifactId>
<version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>3.9.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>kafka</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>io.confluent</groupId>
+ <artifactId>kafka-avro-serializer</artifactId>
+ <version>7.2.2</version>
Review Comment:
Yes, good one!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2174570806
##########
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:
nit: client to connect schema registry -> client to connect to the schema registry
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2139534497
##########
docs/content.zh/docs/connectors/table/formats/avro-confluent.md:
##########
@@ -280,6 +280,13 @@ Format 参数
<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>
Review Comment:
it is worth noting what the default is in the description.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2141951877
##########
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:
Because I don't know Chinese :) So I'll follow the process that's at https://flink.apache.org/how-to-contribute/contribute-documentation/#chinese-documentation-translation after this PR is merged
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2174626983
##########
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:
@fapaul this method implements an interface with a method that throws IOException, I think we should leave it throwing the IOException, so we can use the core interface. I see that the `schemaRegistryClient.getId` throws an IOException, which then comes out of this method. We could wrap the IOException with a new IOException with more descriptive text, I have no strong views to do that. I would leave as coded by @MartijnVisser.
One improvement around the Exception processing would be to put the schema (though this could be big and might need to be truncating) or schema name and subject as inserts in the message. WDYT?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2139541751
##########
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:
is this related to the fix?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2141955041
##########
flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/ConfluentSchemaRegistryCoder.java:
##########
@@ -58,7 +77,7 @@ public ConfluentSchemaRegistryCoder(String subject, SchemaRegistryClient schemaR
* @param schemaRegistryClient client to connect schema registry
*/
public ConfluentSchemaRegistryCoder(SchemaRegistryClient schemaRegistryClient) {
- this.schemaRegistryClient = schemaRegistryClient;
+ this(null, schemaRegistryClient, null);
Review Comment:
That's a valid nit, but the original commit that I took over had that changed already and is in this PR, see https://github.com/apache/flink/blob/0b8f50bc70db8f5eb69e92ca2cf52eb4c83a15fc/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/ConfluentSchemaRegistryCoder.java#L70
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2143064918
##########
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:
I had assumed we would just add here as the optional above is not translated. `<td>yes</td>` as per the note in the link you sent. But it sounds like you have this in hand.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r3960930917
##########
flink-end-to-end-tests/flink-confluent-schema-registry/src/test/java/org/apache/flink/schema/registry/test/AvroConfluentITCase.java:
##########
@@ -0,0 +1,444 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.schema.registry.test;
+
+import org.apache.flink.connector.testframe.container.FlinkContainers;
+import org.apache.flink.connector.testframe.container.FlinkContainersSettings;
+import org.apache.flink.connector.testframe.container.TestcontainersSettings;
+import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.test.resources.ResourceTestUtils;
+import org.apache.flink.test.util.SQLJobSubmission;
+import org.apache.flink.util.DockerImageVersions;
+import org.apache.flink.util.jackson.JacksonMapperFactory;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.apache.avro.generic.GenericRecord;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.Network;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeoutException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** E2E Test for Avro-Confluent integration. */
+@Testcontainers
+public class AvroConfluentITCase {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AvroConfluentITCase.class);
+
+ private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
+ private static final String INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS = "schema-registry";
+
+ private static final String TOPIC = "test-avro-input";
+ private static final String RESULT_TOPIC = "test-avro-output";
+ private static final String MANUAL_TOPIC = "test-avro-input-manual";
+ private static final String MANUAL_RESULT_TOPIC = "test-avro-output-manual";
+
+ private static final String INPUT_SCHEMA = SchemaLoader.loadSchema("avro/input-record.avsc");
+ private static final String OUTPUT_SCHEMA = SchemaLoader.loadSchema("avro/output-record.avsc");
+
+ private static final Path sqlToolBoxJar = ResourceTestUtils.getResource(".*/SqlToolbox\\.jar");
+
+ private final Path sqlConnectorKafkaJar = ResourceTestUtils.getResource(".*kafka.*\\.jar");
+ private final Path sqlConnectorUpsertTestJar =
+ ResourceTestUtils.getResource(".*flink-test-utils.*\\.jar");
+ private final Path sqlAvroConfluentJar =
+ ResourceTestUtils.getResource(".*avro-confluent.*\\.jar");
+
+ private static final Network NETWORK = Network.newNetwork();
+
+ @Container
+ public static final KafkaContainer KAFKA =
+ new KafkaContainer(DockerImageName.parse(DockerImageVersions.KAFKA))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);
+
+ @Container
+ private static final GenericContainer<?> SCHEMA_REGISTRY =
+ new GenericContainer<>(DockerImageName.parse(DockerImageVersions.SCHEMA_REGISTRY))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS)
+ .withExposedPorts(8081)
+ .withEnv("SCHEMA_REGISTRY_HOST_NAME", INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS)
+ .withEnv(
+ "SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS",
+ INTER_CONTAINER_KAFKA_ALIAS + ":9092")
+ .dependsOn(KAFKA);
+
+ private static final FlinkContainers FLINK =
+ FlinkContainers.builder()
+ .withFlinkContainersSettings(
+ FlinkContainersSettings.builder().numTaskManagers(1).build())
+ .withTestcontainersSettings(
+ TestcontainersSettings.builder()
+ .network(NETWORK)
+ .logger(LOG)
+ .dependsOn(KAFKA)
+ .build())
+ .build();
+
+ private final HttpClient client = HttpClient.newHttpClient();
+ private final ObjectMapper objectMapper = JacksonMapperFactory.createObjectMapper();
+
+ @BeforeAll
+ public static void setup() throws Exception {
+ KAFKA.start();
+ SCHEMA_REGISTRY.start();
+ FLINK.start();
+ }
+
+ @AfterAll
+ public static void tearDown() {
+ FLINK.stop();
+ SCHEMA_REGISTRY.stop();
+ KAFKA.stop();
+ }
Review Comment:
Removed, the container annotations and the Flink extension handle the lifecycle now.
##########
flink-end-to-end-tests/flink-confluent-schema-registry/src/test/java/org/apache/flink/schema/registry/test/AvroConfluentITCase.java:
##########
@@ -0,0 +1,444 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.schema.registry.test;
+
+import org.apache.flink.connector.testframe.container.FlinkContainers;
+import org.apache.flink.connector.testframe.container.FlinkContainersSettings;
+import org.apache.flink.connector.testframe.container.TestcontainersSettings;
+import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.test.resources.ResourceTestUtils;
+import org.apache.flink.test.util.SQLJobSubmission;
+import org.apache.flink.util.DockerImageVersions;
+import org.apache.flink.util.jackson.JacksonMapperFactory;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.apache.avro.generic.GenericRecord;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.Network;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeoutException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** E2E Test for Avro-Confluent integration. */
+@Testcontainers
+public class AvroConfluentITCase {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AvroConfluentITCase.class);
+
+ private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
+ private static final String INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS = "schema-registry";
+
+ private static final String TOPIC = "test-avro-input";
+ private static final String RESULT_TOPIC = "test-avro-output";
+ private static final String MANUAL_TOPIC = "test-avro-input-manual";
+ private static final String MANUAL_RESULT_TOPIC = "test-avro-output-manual";
+
+ private static final String INPUT_SCHEMA = SchemaLoader.loadSchema("avro/input-record.avsc");
+ private static final String OUTPUT_SCHEMA = SchemaLoader.loadSchema("avro/output-record.avsc");
+
+ private static final Path sqlToolBoxJar = ResourceTestUtils.getResource(".*/SqlToolbox\\.jar");
+
+ private final Path sqlConnectorKafkaJar = ResourceTestUtils.getResource(".*kafka.*\\.jar");
+ private final Path sqlConnectorUpsertTestJar =
+ ResourceTestUtils.getResource(".*flink-test-utils.*\\.jar");
+ private final Path sqlAvroConfluentJar =
+ ResourceTestUtils.getResource(".*avro-confluent.*\\.jar");
+
+ private static final Network NETWORK = Network.newNetwork();
+
+ @Container
+ public static final KafkaContainer KAFKA =
+ new KafkaContainer(DockerImageName.parse(DockerImageVersions.KAFKA))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);
+
+ @Container
+ private static final GenericContainer<?> SCHEMA_REGISTRY =
+ new GenericContainer<>(DockerImageName.parse(DockerImageVersions.SCHEMA_REGISTRY))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS)
+ .withExposedPorts(8081)
+ .withEnv("SCHEMA_REGISTRY_HOST_NAME", INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS)
+ .withEnv(
+ "SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS",
+ INTER_CONTAINER_KAFKA_ALIAS + ":9092")
+ .dependsOn(KAFKA);
+
+ private static final FlinkContainers FLINK =
+ FlinkContainers.builder()
+ .withFlinkContainersSettings(
+ FlinkContainersSettings.builder().numTaskManagers(1).build())
+ .withTestcontainersSettings(
+ TestcontainersSettings.builder()
+ .network(NETWORK)
+ .logger(LOG)
+ .dependsOn(KAFKA)
+ .build())
+ .build();
+
+ private final HttpClient client = HttpClient.newHttpClient();
+ private final ObjectMapper objectMapper = JacksonMapperFactory.createObjectMapper();
+
+ @BeforeAll
+ public static void setup() throws Exception {
+ KAFKA.start();
+ SCHEMA_REGISTRY.start();
+ FLINK.start();
+ }
+
+ @AfterAll
+ public static void tearDown() {
+ FLINK.stop();
+ SCHEMA_REGISTRY.stop();
+ KAFKA.stop();
+ }
+
+ @BeforeEach
+ public void before() throws Exception {
+ // Create topics using external bootstrap servers since we're outside the Docker network
+ Properties props = new Properties();
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA.getBootstrapServers());
+ try (Admin admin = Admin.create(props)) {
+ admin.createTopics(
+ Arrays.asList(
+ new NewTopic(TOPIC, 1, (short) 1),
+ new NewTopic(RESULT_TOPIC, 1, (short) 1),
+ new NewTopic(MANUAL_TOPIC, 1, (short) 1),
+ new NewTopic(MANUAL_RESULT_TOPIC, 1, (short) 1)));
+
+ // Poll for topic creation with timeout
+ try {
+ CommonTestUtils.waitUntilIgnoringExceptions(
+ () -> {
+ try {
+ Set<String> topics = admin.listTopics().names().get();
+ return topics.contains(TOPIC)
+ && topics.contains(RESULT_TOPIC)
+ && topics.contains(MANUAL_TOPIC)
+ && topics.contains(MANUAL_RESULT_TOPIC);
+ } catch (Exception e) {
+ LOG.warn("Exception while checking topic creation", e);
+ return false;
+ }
+ },
+ Duration.ofSeconds(30),
+ Duration.ofMillis(100),
+ "Topics were not created in time");
+ } catch (TimeoutException | InterruptedException e) {
+ throw new RuntimeException("Failed to wait for topic creation", e);
+ }
+
+ LOG.info(
+ "Topics {}, {}, {}, and {} created successfully",
+ TOPIC,
+ RESULT_TOPIC,
+ MANUAL_TOPIC,
+ MANUAL_RESULT_TOPIC);
+ }
+ }
+
+ @Test
+ public void testAvroConfluentIntegrationWithAutoRegister() throws Exception {
+ // Combine all SQL statements into a single submission
+ List<String> allSqlStatements =
+ Arrays.asList(
+ "SET 'table.dml-sync' = 'true';",
+ "",
+ "CREATE TABLE avro_input (",
+ " name STRING,",
+ " favoriteNumber STRING,",
+ " favoriteColor STRING,",
+ " eventType STRING",
+ ") WITH (",
+ " 'connector' = 'kafka',",
+ " 'topic' = '" + TOPIC + "',",
+ " 'properties.bootstrap.servers' = '"
+ + INTER_CONTAINER_KAFKA_ALIAS
+ + ":9092',",
+ " 'properties.group.id' = 'test-group',",
+ " 'scan.startup.mode' = 'earliest-offset',",
+ " 'scan.bounded.mode' = 'latest-offset',",
+ " 'format' = 'avro-confluent',",
+ " 'avro-confluent.url' = 'http://"
+ + INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS
+ + ":8081',",
+ " 'avro-confluent.auto.register.schemas' = 'true'",
+ ");",
+ "",
+ "CREATE TABLE avro_output (",
+ " name STRING,",
+ " favoriteNumber STRING,",
+ " favoriteColor STRING,",
+ " eventType STRING",
+ ") WITH (",
+ " 'connector' = 'kafka',",
+ " 'topic' = '" + RESULT_TOPIC + "',",
+ " 'properties.bootstrap.servers' = '"
+ + INTER_CONTAINER_KAFKA_ALIAS
+ + ":9092',",
+ " 'properties.group.id' = 'test-group',",
+ " 'scan.startup.mode' = 'earliest-offset',",
+ " 'scan.bounded.mode' = 'latest-offset',",
+ " 'format' = 'avro-confluent',",
+ " 'avro-confluent.url' = 'http://"
+ + INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS
+ + ":8081',",
+ " 'avro-confluent.auto.register.schemas' = 'true'",
+ ");",
+ "",
+ "INSERT INTO avro_input VALUES",
+ " ('Alice', '42', 'blue', 'INSERT'),",
+ " ('Bob', '7', 'red', 'INSERT'),",
+ " ('Charlie', '73', 'green', 'INSERT');",
+ "",
+ "INSERT INTO avro_output",
+ "SELECT * FROM avro_input;");
+
+ LOG.info("Submitting SQL statements: {}", String.join("\n", allSqlStatements));
+
+ // Execute all SQL statements in a single submission
+ executeSql(allSqlStatements);
+
+ // Verify output
+ verifyNumberOfResultRecords(RESULT_TOPIC, 3);
+ }
+
+ @Test
+ public void testAvroConfluentIntegrationWithManualRegister() throws Exception {
Review Comment:
Added `testWritingFailsWhenAutoRegisterDisabledAndSchemaMissing`. It writes to a topic without a registered subject, asserts the job ends in FAILED and that no subject was created afterwards. The failure is the 40401 from the registry, wrapped in the new error message.
##########
flink-end-to-end-tests/flink-confluent-schema-registry/src/test/java/org/apache/flink/schema/registry/test/AvroConfluentITCase.java:
##########
@@ -0,0 +1,444 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.schema.registry.test;
+
+import org.apache.flink.connector.testframe.container.FlinkContainers;
+import org.apache.flink.connector.testframe.container.FlinkContainersSettings;
+import org.apache.flink.connector.testframe.container.TestcontainersSettings;
+import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.test.resources.ResourceTestUtils;
+import org.apache.flink.test.util.SQLJobSubmission;
+import org.apache.flink.util.DockerImageVersions;
+import org.apache.flink.util.jackson.JacksonMapperFactory;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.apache.avro.generic.GenericRecord;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.Network;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeoutException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** E2E Test for Avro-Confluent integration. */
+@Testcontainers
+public class AvroConfluentITCase {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AvroConfluentITCase.class);
+
+ private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
+ private static final String INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS = "schema-registry";
+
+ private static final String TOPIC = "test-avro-input";
+ private static final String RESULT_TOPIC = "test-avro-output";
+ private static final String MANUAL_TOPIC = "test-avro-input-manual";
+ private static final String MANUAL_RESULT_TOPIC = "test-avro-output-manual";
+
+ private static final String INPUT_SCHEMA = SchemaLoader.loadSchema("avro/input-record.avsc");
+ private static final String OUTPUT_SCHEMA = SchemaLoader.loadSchema("avro/output-record.avsc");
+
+ private static final Path sqlToolBoxJar = ResourceTestUtils.getResource(".*/SqlToolbox\\.jar");
+
+ private final Path sqlConnectorKafkaJar = ResourceTestUtils.getResource(".*kafka.*\\.jar");
+ private final Path sqlConnectorUpsertTestJar =
+ ResourceTestUtils.getResource(".*flink-test-utils.*\\.jar");
+ private final Path sqlAvroConfluentJar =
+ ResourceTestUtils.getResource(".*avro-confluent.*\\.jar");
+
+ private static final Network NETWORK = Network.newNetwork();
+
+ @Container
+ public static final KafkaContainer KAFKA =
+ new KafkaContainer(DockerImageName.parse(DockerImageVersions.KAFKA))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);
+
+ @Container
+ private static final GenericContainer<?> SCHEMA_REGISTRY =
+ new GenericContainer<>(DockerImageName.parse(DockerImageVersions.SCHEMA_REGISTRY))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS)
+ .withExposedPorts(8081)
+ .withEnv("SCHEMA_REGISTRY_HOST_NAME", INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS)
+ .withEnv(
+ "SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS",
+ INTER_CONTAINER_KAFKA_ALIAS + ":9092")
+ .dependsOn(KAFKA);
+
+ private static final FlinkContainers FLINK =
+ FlinkContainers.builder()
+ .withFlinkContainersSettings(
+ FlinkContainersSettings.builder().numTaskManagers(1).build())
+ .withTestcontainersSettings(
+ TestcontainersSettings.builder()
+ .network(NETWORK)
+ .logger(LOG)
+ .dependsOn(KAFKA)
+ .build())
+ .build();
+
+ private final HttpClient client = HttpClient.newHttpClient();
+ private final ObjectMapper objectMapper = JacksonMapperFactory.createObjectMapper();
+
+ @BeforeAll
+ public static void setup() throws Exception {
+ KAFKA.start();
+ SCHEMA_REGISTRY.start();
+ FLINK.start();
+ }
+
+ @AfterAll
+ public static void tearDown() {
+ FLINK.stop();
+ SCHEMA_REGISTRY.stop();
+ KAFKA.stop();
+ }
+
+ @BeforeEach
+ public void before() throws Exception {
+ // Create topics using external bootstrap servers since we're outside the Docker network
+ Properties props = new Properties();
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA.getBootstrapServers());
+ try (Admin admin = Admin.create(props)) {
+ admin.createTopics(
+ Arrays.asList(
+ new NewTopic(TOPIC, 1, (short) 1),
+ new NewTopic(RESULT_TOPIC, 1, (short) 1),
+ new NewTopic(MANUAL_TOPIC, 1, (short) 1),
+ new NewTopic(MANUAL_RESULT_TOPIC, 1, (short) 1)));
+
+ // Poll for topic creation with timeout
+ try {
+ CommonTestUtils.waitUntilIgnoringExceptions(
+ () -> {
+ try {
+ Set<String> topics = admin.listTopics().names().get();
+ return topics.contains(TOPIC)
+ && topics.contains(RESULT_TOPIC)
+ && topics.contains(MANUAL_TOPIC)
+ && topics.contains(MANUAL_RESULT_TOPIC);
+ } catch (Exception e) {
+ LOG.warn("Exception while checking topic creation", e);
+ return false;
+ }
+ },
+ Duration.ofSeconds(30),
+ Duration.ofMillis(100),
+ "Topics were not created in time");
+ } catch (TimeoutException | InterruptedException e) {
+ throw new RuntimeException("Failed to wait for topic creation", e);
+ }
+
+ LOG.info(
+ "Topics {}, {}, {}, and {} created successfully",
+ TOPIC,
+ RESULT_TOPIC,
+ MANUAL_TOPIC,
+ MANUAL_RESULT_TOPIC);
+ }
+ }
+
+ @Test
+ public void testAvroConfluentIntegrationWithAutoRegister() throws Exception {
+ // Combine all SQL statements into a single submission
+ List<String> allSqlStatements =
+ Arrays.asList(
+ "SET 'table.dml-sync' = 'true';",
+ "",
+ "CREATE TABLE avro_input (",
+ " name STRING,",
+ " favoriteNumber STRING,",
+ " favoriteColor STRING,",
+ " eventType STRING",
+ ") WITH (",
+ " 'connector' = 'kafka',",
+ " 'topic' = '" + TOPIC + "',",
+ " 'properties.bootstrap.servers' = '"
+ + INTER_CONTAINER_KAFKA_ALIAS
+ + ":9092',",
+ " 'properties.group.id' = 'test-group',",
+ " 'scan.startup.mode' = 'earliest-offset',",
+ " 'scan.bounded.mode' = 'latest-offset',",
+ " 'format' = 'avro-confluent',",
+ " 'avro-confluent.url' = 'http://"
+ + INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS
+ + ":8081',",
+ " 'avro-confluent.auto.register.schemas' = 'true'",
+ ");",
+ "",
+ "CREATE TABLE avro_output (",
+ " name STRING,",
+ " favoriteNumber STRING,",
+ " favoriteColor STRING,",
+ " eventType STRING",
+ ") WITH (",
+ " 'connector' = 'kafka',",
+ " 'topic' = '" + RESULT_TOPIC + "',",
+ " 'properties.bootstrap.servers' = '"
+ + INTER_CONTAINER_KAFKA_ALIAS
+ + ":9092',",
+ " 'properties.group.id' = 'test-group',",
+ " 'scan.startup.mode' = 'earliest-offset',",
+ " 'scan.bounded.mode' = 'latest-offset',",
+ " 'format' = 'avro-confluent',",
+ " 'avro-confluent.url' = 'http://"
+ + INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS
+ + ":8081',",
+ " 'avro-confluent.auto.register.schemas' = 'true'",
+ ");",
+ "",
+ "INSERT INTO avro_input VALUES",
+ " ('Alice', '42', 'blue', 'INSERT'),",
+ " ('Bob', '7', 'red', 'INSERT'),",
+ " ('Charlie', '73', 'green', 'INSERT');",
+ "",
+ "INSERT INTO avro_output",
+ "SELECT * FROM avro_input;");
Review Comment:
Done, both round-trip tests now use `avro_confluent_roundtrip_e2e.sql` with the flag and topics as variables.
##########
flink-end-to-end-tests/flink-confluent-schema-registry/src/test/java/org/apache/flink/schema/registry/test/AvroConfluentITCase.java:
##########
@@ -0,0 +1,444 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.schema.registry.test;
+
+import org.apache.flink.connector.testframe.container.FlinkContainers;
+import org.apache.flink.connector.testframe.container.FlinkContainersSettings;
+import org.apache.flink.connector.testframe.container.TestcontainersSettings;
+import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.test.resources.ResourceTestUtils;
+import org.apache.flink.test.util.SQLJobSubmission;
+import org.apache.flink.util.DockerImageVersions;
+import org.apache.flink.util.jackson.JacksonMapperFactory;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.apache.avro.generic.GenericRecord;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.Network;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeoutException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** E2E Test for Avro-Confluent integration. */
+@Testcontainers
+public class AvroConfluentITCase {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AvroConfluentITCase.class);
+
+ private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
+ private static final String INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS = "schema-registry";
+
+ private static final String TOPIC = "test-avro-input";
+ private static final String RESULT_TOPIC = "test-avro-output";
+ private static final String MANUAL_TOPIC = "test-avro-input-manual";
+ private static final String MANUAL_RESULT_TOPIC = "test-avro-output-manual";
+
+ private static final String INPUT_SCHEMA = SchemaLoader.loadSchema("avro/input-record.avsc");
+ private static final String OUTPUT_SCHEMA = SchemaLoader.loadSchema("avro/output-record.avsc");
+
+ private static final Path sqlToolBoxJar = ResourceTestUtils.getResource(".*/SqlToolbox\\.jar");
+
+ private final Path sqlConnectorKafkaJar = ResourceTestUtils.getResource(".*kafka.*\\.jar");
+ private final Path sqlConnectorUpsertTestJar =
+ ResourceTestUtils.getResource(".*flink-test-utils.*\\.jar");
+ private final Path sqlAvroConfluentJar =
+ ResourceTestUtils.getResource(".*avro-confluent.*\\.jar");
+
+ private static final Network NETWORK = Network.newNetwork();
+
+ @Container
+ public static final KafkaContainer KAFKA =
+ new KafkaContainer(DockerImageName.parse(DockerImageVersions.KAFKA))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);
+
+ @Container
+ private static final GenericContainer<?> SCHEMA_REGISTRY =
+ new GenericContainer<>(DockerImageName.parse(DockerImageVersions.SCHEMA_REGISTRY))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS)
+ .withExposedPorts(8081)
+ .withEnv("SCHEMA_REGISTRY_HOST_NAME", INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS)
+ .withEnv(
+ "SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS",
+ INTER_CONTAINER_KAFKA_ALIAS + ":9092")
+ .dependsOn(KAFKA);
+
+ private static final FlinkContainers FLINK =
+ FlinkContainers.builder()
+ .withFlinkContainersSettings(
+ FlinkContainersSettings.builder().numTaskManagers(1).build())
+ .withTestcontainersSettings(
+ TestcontainersSettings.builder()
+ .network(NETWORK)
+ .logger(LOG)
+ .dependsOn(KAFKA)
+ .build())
+ .build();
+
+ private final HttpClient client = HttpClient.newHttpClient();
+ private final ObjectMapper objectMapper = JacksonMapperFactory.createObjectMapper();
Review Comment:
Done.
##########
docs/content.zh/docs/connectors/table/formats/avro-confluent.md:
##########
@@ -280,6 +280,13 @@ Format 参数
<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 style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether to automatically register schemas with the Confluent Schema Registry if they don't exist. When set to <code>false</code>, schemas must be manually registered in the Schema Registry before being used. When set to <code>true</code>, schemas will be automatically registered during serialization if they don't already exist. The default value is <code>true</code>.</td>
Review Comment:
Added to the format description and to the option text: registration only happens when writing, reading always looks up the writer schema by the id in the record.
##########
flink-end-to-end-tests/flink-confluent-schema-registry/src/test/resources/log4j2-test.properties:
##########
@@ -0,0 +1,34 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+# Set root logger level to OFF to not flood build logs
+# set manually to INFO for debugging purposes
+rootLogger.level = INFO
Review Comment:
Done.
##########
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);
+ }
}
+ out.write(CONFLUENT_MAGIC_BYTE);
+ byte[] schemaIdBytes = ByteBuffer.allocate(4).putInt(registeredId).array();
+ out.write(schemaIdBytes);
+ }
+
+ private boolean registerSchema() {
Review Comment:
It's parsed once in the constructor now, including a check that rejects anything other than true/false.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2139526384
##########
flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/ConfluentSchemaRegistryCoder.java:
##########
@@ -58,7 +77,7 @@ public ConfluentSchemaRegistryCoder(String subject, SchemaRegistryClient schemaR
* @param schemaRegistryClient client to connect schema registry
*/
public ConfluentSchemaRegistryCoder(SchemaRegistryClient schemaRegistryClient) {
- this.schemaRegistryClient = schemaRegistryClient;
+ this(null, schemaRegistryClient, null);
Review Comment:
nit: it is a bit strange to have an optional first parameter I would expect the mandatory parameters first for all these methods ie the constructors to all start
`ConfluentSchemaRegistryCoder(SchemaRegistryClient schemaRegistryClient` ...
I realise this is existing code, so it is your call if you want to change it
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2139538705
##########
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:
why is this line here but not in the Chinese?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2172024538
##########
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);
Review Comment:
I was expecting the code to try to get the schema in all cases , then if there is no schema found we check the config to see if we need to register it otherwise fail. This would be in line with the docs "Whether to automatically register schemas with the Confluent Schema Registry if they don't exist."
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "raphaelauv (via GitHub)" <gi...@apache.org>.
raphaelauv commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2282791314
##########
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:
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 ?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on PR #26662:
URL: https://github.com/apache/flink/pull/26662#issuecomment-5589790704
@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.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2141956278
##########
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:
It's about keeping versions in sync. Ideally the Python dependencies on connector versions should be out of the Flink repo and moved into the connector repos, but we're not there yet
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "fapaul (via GitHub)" <gi...@apache.org>.
fapaul commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2158361081
##########
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:
Nit: Why did you change the used type of the example? Is this related to registration?
##########
flink-end-to-end-tests/flink-confluent-schema-registry/src/test/java/org/apache/flink/schema/registry/test/AvroConfluentITCase.java:
##########
@@ -0,0 +1,444 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.schema.registry.test;
+
+import org.apache.flink.connector.testframe.container.FlinkContainers;
+import org.apache.flink.connector.testframe.container.FlinkContainersSettings;
+import org.apache.flink.connector.testframe.container.TestcontainersSettings;
+import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.test.resources.ResourceTestUtils;
+import org.apache.flink.test.util.SQLJobSubmission;
+import org.apache.flink.util.DockerImageVersions;
+import org.apache.flink.util.jackson.JacksonMapperFactory;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.apache.avro.generic.GenericRecord;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.Network;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeoutException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** E2E Test for Avro-Confluent integration. */
+@Testcontainers
+public class AvroConfluentITCase {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AvroConfluentITCase.class);
+
+ private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
+ private static final String INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS = "schema-registry";
+
+ private static final String TOPIC = "test-avro-input";
+ private static final String RESULT_TOPIC = "test-avro-output";
+ private static final String MANUAL_TOPIC = "test-avro-input-manual";
+ private static final String MANUAL_RESULT_TOPIC = "test-avro-output-manual";
+
+ private static final String INPUT_SCHEMA = SchemaLoader.loadSchema("avro/input-record.avsc");
+ private static final String OUTPUT_SCHEMA = SchemaLoader.loadSchema("avro/output-record.avsc");
+
+ private static final Path sqlToolBoxJar = ResourceTestUtils.getResource(".*/SqlToolbox\\.jar");
+
+ private final Path sqlConnectorKafkaJar = ResourceTestUtils.getResource(".*kafka.*\\.jar");
+ private final Path sqlConnectorUpsertTestJar =
+ ResourceTestUtils.getResource(".*flink-test-utils.*\\.jar");
+ private final Path sqlAvroConfluentJar =
+ ResourceTestUtils.getResource(".*avro-confluent.*\\.jar");
+
+ private static final Network NETWORK = Network.newNetwork();
+
+ @Container
+ public static final KafkaContainer KAFKA =
+ new KafkaContainer(DockerImageName.parse(DockerImageVersions.KAFKA))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);
+
+ @Container
+ private static final GenericContainer<?> SCHEMA_REGISTRY =
+ new GenericContainer<>(DockerImageName.parse(DockerImageVersions.SCHEMA_REGISTRY))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS)
+ .withExposedPorts(8081)
+ .withEnv("SCHEMA_REGISTRY_HOST_NAME", INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS)
+ .withEnv(
+ "SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS",
+ INTER_CONTAINER_KAFKA_ALIAS + ":9092")
+ .dependsOn(KAFKA);
+
+ private static final FlinkContainers FLINK =
+ FlinkContainers.builder()
+ .withFlinkContainersSettings(
+ FlinkContainersSettings.builder().numTaskManagers(1).build())
+ .withTestcontainersSettings(
+ TestcontainersSettings.builder()
+ .network(NETWORK)
+ .logger(LOG)
+ .dependsOn(KAFKA)
+ .build())
+ .build();
+
+ private final HttpClient client = HttpClient.newHttpClient();
+ private final ObjectMapper objectMapper = JacksonMapperFactory.createObjectMapper();
+
+ @BeforeAll
+ public static void setup() throws Exception {
+ KAFKA.start();
+ SCHEMA_REGISTRY.start();
+ FLINK.start();
+ }
+
+ @AfterAll
+ public static void tearDown() {
+ FLINK.stop();
+ SCHEMA_REGISTRY.stop();
+ KAFKA.stop();
+ }
+
+ @BeforeEach
+ public void before() throws Exception {
+ // Create topics using external bootstrap servers since we're outside the Docker network
+ Properties props = new Properties();
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA.getBootstrapServers());
+ try (Admin admin = Admin.create(props)) {
+ admin.createTopics(
+ Arrays.asList(
+ new NewTopic(TOPIC, 1, (short) 1),
+ new NewTopic(RESULT_TOPIC, 1, (short) 1),
+ new NewTopic(MANUAL_TOPIC, 1, (short) 1),
+ new NewTopic(MANUAL_RESULT_TOPIC, 1, (short) 1)));
+
+ // Poll for topic creation with timeout
+ try {
+ CommonTestUtils.waitUntilIgnoringExceptions(
+ () -> {
+ try {
+ Set<String> topics = admin.listTopics().names().get();
+ return topics.contains(TOPIC)
+ && topics.contains(RESULT_TOPIC)
+ && topics.contains(MANUAL_TOPIC)
+ && topics.contains(MANUAL_RESULT_TOPIC);
+ } catch (Exception e) {
+ LOG.warn("Exception while checking topic creation", e);
+ return false;
+ }
+ },
+ Duration.ofSeconds(30),
+ Duration.ofMillis(100),
+ "Topics were not created in time");
+ } catch (TimeoutException | InterruptedException e) {
+ throw new RuntimeException("Failed to wait for topic creation", e);
+ }
+
+ LOG.info(
+ "Topics {}, {}, {}, and {} created successfully",
+ TOPIC,
+ RESULT_TOPIC,
+ MANUAL_TOPIC,
+ MANUAL_RESULT_TOPIC);
+ }
+ }
+
+ @Test
+ public void testAvroConfluentIntegrationWithAutoRegister() throws Exception {
+ // Combine all SQL statements into a single submission
+ List<String> allSqlStatements =
+ Arrays.asList(
+ "SET 'table.dml-sync' = 'true';",
+ "",
+ "CREATE TABLE avro_input (",
+ " name STRING,",
+ " favoriteNumber STRING,",
+ " favoriteColor STRING,",
+ " eventType STRING",
+ ") WITH (",
+ " 'connector' = 'kafka',",
+ " 'topic' = '" + TOPIC + "',",
+ " 'properties.bootstrap.servers' = '"
+ + INTER_CONTAINER_KAFKA_ALIAS
+ + ":9092',",
+ " 'properties.group.id' = 'test-group',",
+ " 'scan.startup.mode' = 'earliest-offset',",
+ " 'scan.bounded.mode' = 'latest-offset',",
+ " 'format' = 'avro-confluent',",
+ " 'avro-confluent.url' = 'http://"
+ + INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS
+ + ":8081',",
+ " 'avro-confluent.auto.register.schemas' = 'true'",
+ ");",
+ "",
+ "CREATE TABLE avro_output (",
+ " name STRING,",
+ " favoriteNumber STRING,",
+ " favoriteColor STRING,",
+ " eventType STRING",
+ ") WITH (",
+ " 'connector' = 'kafka',",
+ " 'topic' = '" + RESULT_TOPIC + "',",
+ " 'properties.bootstrap.servers' = '"
+ + INTER_CONTAINER_KAFKA_ALIAS
+ + ":9092',",
+ " 'properties.group.id' = 'test-group',",
+ " 'scan.startup.mode' = 'earliest-offset',",
+ " 'scan.bounded.mode' = 'latest-offset',",
+ " 'format' = 'avro-confluent',",
+ " 'avro-confluent.url' = 'http://"
+ + INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS
+ + ":8081',",
+ " 'avro-confluent.auto.register.schemas' = 'true'",
+ ");",
+ "",
+ "INSERT INTO avro_input VALUES",
+ " ('Alice', '42', 'blue', 'INSERT'),",
+ " ('Bob', '7', 'red', 'INSERT'),",
+ " ('Charlie', '73', 'green', 'INSERT');",
+ "",
+ "INSERT INTO avro_output",
+ "SELECT * FROM avro_input;");
+
+ LOG.info("Submitting SQL statements: {}", String.join("\n", allSqlStatements));
+
+ // Execute all SQL statements in a single submission
+ executeSql(allSqlStatements);
+
+ // Verify output
+ verifyNumberOfResultRecords(RESULT_TOPIC, 3);
+ }
+
+ @Test
+ public void testAvroConfluentIntegrationWithManualRegister() throws Exception {
Review Comment:
I would have expected to see a test that runs a query and fails, if the schema isn't registered.
The current test IMO doesn't fully cover the behavior since it can also pass if the SQL query does the registration.
##########
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);
+ }
}
+ out.write(CONFLUENT_MAGIC_BYTE);
+ byte[] schemaIdBytes = ByteBuffer.allocate(4).putInt(registeredId).array();
+ out.write(schemaIdBytes);
+ }
+
+ private boolean registerSchema() {
Review Comment:
Why is `registerSchema` a method, instead of parsing the value once in the ctor?
##########
flink-end-to-end-tests/flink-confluent-schema-registry/pom.xml:
##########
@@ -43,101 +43,160 @@ under the License.
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-streaming-java</artifactId>
+ <artifactId>flink-table-common</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
- <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka -->
+
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-kafka</artifactId>
- <version>3.0.0-1.17</version>
+ <artifactId>flink-table-api-java-bridge</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
</dependency>
- <!-- Make sure that Shaded Guava matches the one used in the flink-connector-kafka,
- or remove when FLINK-32462 is resolved -->
+
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-shaded-guava</artifactId>
- <version>30.1.1-jre-16.1</version>
+ <artifactId>flink-end-to-end-tests-common</artifactId>
+ <version>${project.version}</version>
</dependency>
+
+ <!-- This enables the WebUI during tests. -->
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-avro</artifactId>
+ <artifactId>flink-runtime-web</artifactId>
<version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- The following dependencies are for connector/format sql-jars that
+ we copy using the maven-dependency-plugin. When extending the test
+ to cover more connectors/formats, add a dependency here and an entry
+ to the dependency-plugin configuration below.
+ This ensures that all modules we actually need (as defined by the
+ dependency-plugin configuration) are built before this module. -->
+ <dependency>
+ <!-- Used by maven-dependency-plugin -->
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-sql-connector-kafka</artifactId>
+ <version>4.0.0-2.0</version>
+ <scope>provided</scope>
</dependency>
+
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-avro-confluent-registry</artifactId>
+ <artifactId>flink-sql-avro-confluent-registry</artifactId>
<version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>3.9.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>kafka</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>io.confluent</groupId>
+ <artifactId>kafka-avro-serializer</artifactId>
+ <version>7.2.2</version>
Review Comment:
Afaik the serializer is versioned similarly to kafka. 7.2.2 should be Kafka 3.2. Can we upgrade the serializer to `7.9.0` to be inline with the used kafka version 3.9.0?
##########
flink-end-to-end-tests/flink-confluent-schema-registry/src/test/java/org/apache/flink/schema/registry/test/AvroConfluentITCase.java:
##########
@@ -0,0 +1,444 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.schema.registry.test;
+
+import org.apache.flink.connector.testframe.container.FlinkContainers;
+import org.apache.flink.connector.testframe.container.FlinkContainersSettings;
+import org.apache.flink.connector.testframe.container.TestcontainersSettings;
+import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.test.resources.ResourceTestUtils;
+import org.apache.flink.test.util.SQLJobSubmission;
+import org.apache.flink.util.DockerImageVersions;
+import org.apache.flink.util.jackson.JacksonMapperFactory;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.apache.avro.generic.GenericRecord;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.Network;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeoutException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** E2E Test for Avro-Confluent integration. */
+@Testcontainers
+public class AvroConfluentITCase {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AvroConfluentITCase.class);
+
+ private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
+ private static final String INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS = "schema-registry";
+
+ private static final String TOPIC = "test-avro-input";
+ private static final String RESULT_TOPIC = "test-avro-output";
+ private static final String MANUAL_TOPIC = "test-avro-input-manual";
+ private static final String MANUAL_RESULT_TOPIC = "test-avro-output-manual";
+
+ private static final String INPUT_SCHEMA = SchemaLoader.loadSchema("avro/input-record.avsc");
+ private static final String OUTPUT_SCHEMA = SchemaLoader.loadSchema("avro/output-record.avsc");
+
+ private static final Path sqlToolBoxJar = ResourceTestUtils.getResource(".*/SqlToolbox\\.jar");
+
+ private final Path sqlConnectorKafkaJar = ResourceTestUtils.getResource(".*kafka.*\\.jar");
+ private final Path sqlConnectorUpsertTestJar =
+ ResourceTestUtils.getResource(".*flink-test-utils.*\\.jar");
+ private final Path sqlAvroConfluentJar =
+ ResourceTestUtils.getResource(".*avro-confluent.*\\.jar");
+
+ private static final Network NETWORK = Network.newNetwork();
+
+ @Container
+ public static final KafkaContainer KAFKA =
+ new KafkaContainer(DockerImageName.parse(DockerImageVersions.KAFKA))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);
+
+ @Container
+ private static final GenericContainer<?> SCHEMA_REGISTRY =
+ new GenericContainer<>(DockerImageName.parse(DockerImageVersions.SCHEMA_REGISTRY))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS)
+ .withExposedPorts(8081)
+ .withEnv("SCHEMA_REGISTRY_HOST_NAME", INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS)
+ .withEnv(
+ "SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS",
+ INTER_CONTAINER_KAFKA_ALIAS + ":9092")
+ .dependsOn(KAFKA);
+
+ private static final FlinkContainers FLINK =
+ FlinkContainers.builder()
+ .withFlinkContainersSettings(
+ FlinkContainersSettings.builder().numTaskManagers(1).build())
+ .withTestcontainersSettings(
+ TestcontainersSettings.builder()
+ .network(NETWORK)
+ .logger(LOG)
+ .dependsOn(KAFKA)
+ .build())
+ .build();
+
+ private final HttpClient client = HttpClient.newHttpClient();
+ private final ObjectMapper objectMapper = JacksonMapperFactory.createObjectMapper();
+
+ @BeforeAll
+ public static void setup() throws Exception {
+ KAFKA.start();
+ SCHEMA_REGISTRY.start();
+ FLINK.start();
+ }
+
+ @AfterAll
+ public static void tearDown() {
+ FLINK.stop();
+ SCHEMA_REGISTRY.stop();
+ KAFKA.stop();
+ }
+
+ @BeforeEach
+ public void before() throws Exception {
+ // Create topics using external bootstrap servers since we're outside the Docker network
+ Properties props = new Properties();
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA.getBootstrapServers());
+ try (Admin admin = Admin.create(props)) {
+ admin.createTopics(
+ Arrays.asList(
+ new NewTopic(TOPIC, 1, (short) 1),
+ new NewTopic(RESULT_TOPIC, 1, (short) 1),
+ new NewTopic(MANUAL_TOPIC, 1, (short) 1),
+ new NewTopic(MANUAL_RESULT_TOPIC, 1, (short) 1)));
+
+ // Poll for topic creation with timeout
+ try {
+ CommonTestUtils.waitUntilIgnoringExceptions(
+ () -> {
+ try {
+ Set<String> topics = admin.listTopics().names().get();
+ return topics.contains(TOPIC)
+ && topics.contains(RESULT_TOPIC)
+ && topics.contains(MANUAL_TOPIC)
+ && topics.contains(MANUAL_RESULT_TOPIC);
+ } catch (Exception e) {
+ LOG.warn("Exception while checking topic creation", e);
+ return false;
+ }
+ },
+ Duration.ofSeconds(30),
+ Duration.ofMillis(100),
+ "Topics were not created in time");
+ } catch (TimeoutException | InterruptedException e) {
+ throw new RuntimeException("Failed to wait for topic creation", e);
+ }
+
+ LOG.info(
+ "Topics {}, {}, {}, and {} created successfully",
+ TOPIC,
+ RESULT_TOPIC,
+ MANUAL_TOPIC,
+ MANUAL_RESULT_TOPIC);
+ }
+ }
+
+ @Test
+ public void testAvroConfluentIntegrationWithAutoRegister() throws Exception {
+ // Combine all SQL statements into a single submission
+ List<String> allSqlStatements =
+ Arrays.asList(
+ "SET 'table.dml-sync' = 'true';",
+ "",
+ "CREATE TABLE avro_input (",
+ " name STRING,",
+ " favoriteNumber STRING,",
+ " favoriteColor STRING,",
+ " eventType STRING",
+ ") WITH (",
+ " 'connector' = 'kafka',",
+ " 'topic' = '" + TOPIC + "',",
+ " 'properties.bootstrap.servers' = '"
+ + INTER_CONTAINER_KAFKA_ALIAS
+ + ":9092',",
+ " 'properties.group.id' = 'test-group',",
+ " 'scan.startup.mode' = 'earliest-offset',",
+ " 'scan.bounded.mode' = 'latest-offset',",
+ " 'format' = 'avro-confluent',",
+ " 'avro-confluent.url' = 'http://"
+ + INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS
+ + ":8081',",
+ " 'avro-confluent.auto.register.schemas' = 'true'",
+ ");",
+ "",
+ "CREATE TABLE avro_output (",
+ " name STRING,",
+ " favoriteNumber STRING,",
+ " favoriteColor STRING,",
+ " eventType STRING",
+ ") WITH (",
+ " 'connector' = 'kafka',",
+ " 'topic' = '" + RESULT_TOPIC + "',",
+ " 'properties.bootstrap.servers' = '"
+ + INTER_CONTAINER_KAFKA_ALIAS
+ + ":9092',",
+ " 'properties.group.id' = 'test-group',",
+ " 'scan.startup.mode' = 'earliest-offset',",
+ " 'scan.bounded.mode' = 'latest-offset',",
+ " 'format' = 'avro-confluent',",
+ " 'avro-confluent.url' = 'http://"
+ + INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS
+ + ":8081',",
+ " 'avro-confluent.auto.register.schemas' = 'true'",
+ ");",
+ "",
+ "INSERT INTO avro_input VALUES",
+ " ('Alice', '42', 'blue', 'INSERT'),",
+ " ('Bob', '7', 'red', 'INSERT'),",
+ " ('Charlie', '73', 'green', 'INSERT');",
+ "",
+ "INSERT INTO avro_output",
+ "SELECT * FROM avro_input;");
Review Comment:
Nit: Can you extract the String to a template? Currently, it's hard to review and spot the difference between the two test cases.
##########
flink-end-to-end-tests/flink-confluent-schema-registry/src/test/java/org/apache/flink/schema/registry/test/AvroConfluentITCase.java:
##########
@@ -0,0 +1,444 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.schema.registry.test;
+
+import org.apache.flink.connector.testframe.container.FlinkContainers;
+import org.apache.flink.connector.testframe.container.FlinkContainersSettings;
+import org.apache.flink.connector.testframe.container.TestcontainersSettings;
+import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.test.resources.ResourceTestUtils;
+import org.apache.flink.test.util.SQLJobSubmission;
+import org.apache.flink.util.DockerImageVersions;
+import org.apache.flink.util.jackson.JacksonMapperFactory;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.apache.avro.generic.GenericRecord;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.Network;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeoutException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** E2E Test for Avro-Confluent integration. */
+@Testcontainers
+public class AvroConfluentITCase {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AvroConfluentITCase.class);
+
+ private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
+ private static final String INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS = "schema-registry";
+
+ private static final String TOPIC = "test-avro-input";
+ private static final String RESULT_TOPIC = "test-avro-output";
+ private static final String MANUAL_TOPIC = "test-avro-input-manual";
+ private static final String MANUAL_RESULT_TOPIC = "test-avro-output-manual";
+
+ private static final String INPUT_SCHEMA = SchemaLoader.loadSchema("avro/input-record.avsc");
+ private static final String OUTPUT_SCHEMA = SchemaLoader.loadSchema("avro/output-record.avsc");
+
+ private static final Path sqlToolBoxJar = ResourceTestUtils.getResource(".*/SqlToolbox\\.jar");
+
+ private final Path sqlConnectorKafkaJar = ResourceTestUtils.getResource(".*kafka.*\\.jar");
+ private final Path sqlConnectorUpsertTestJar =
+ ResourceTestUtils.getResource(".*flink-test-utils.*\\.jar");
+ private final Path sqlAvroConfluentJar =
+ ResourceTestUtils.getResource(".*avro-confluent.*\\.jar");
+
+ private static final Network NETWORK = Network.newNetwork();
+
+ @Container
+ public static final KafkaContainer KAFKA =
+ new KafkaContainer(DockerImageName.parse(DockerImageVersions.KAFKA))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);
+
+ @Container
+ private static final GenericContainer<?> SCHEMA_REGISTRY =
+ new GenericContainer<>(DockerImageName.parse(DockerImageVersions.SCHEMA_REGISTRY))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS)
+ .withExposedPorts(8081)
+ .withEnv("SCHEMA_REGISTRY_HOST_NAME", INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS)
+ .withEnv(
+ "SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS",
+ INTER_CONTAINER_KAFKA_ALIAS + ":9092")
+ .dependsOn(KAFKA);
+
+ private static final FlinkContainers FLINK =
+ FlinkContainers.builder()
+ .withFlinkContainersSettings(
+ FlinkContainersSettings.builder().numTaskManagers(1).build())
+ .withTestcontainersSettings(
+ TestcontainersSettings.builder()
+ .network(NETWORK)
+ .logger(LOG)
+ .dependsOn(KAFKA)
+ .build())
+ .build();
+
+ private final HttpClient client = HttpClient.newHttpClient();
+ private final ObjectMapper objectMapper = JacksonMapperFactory.createObjectMapper();
+
+ @BeforeAll
+ public static void setup() throws Exception {
+ KAFKA.start();
+ SCHEMA_REGISTRY.start();
+ FLINK.start();
+ }
+
+ @AfterAll
+ public static void tearDown() {
+ FLINK.stop();
+ SCHEMA_REGISTRY.stop();
+ KAFKA.stop();
+ }
Review Comment:
These methods should be obsolete since for the Kafka/SR container the lifecycle is already handled by using the `@Container` annotation and for the Flink container see the commend above.
##########
flink-end-to-end-tests/flink-confluent-schema-registry/src/test/java/org/apache/flink/schema/registry/test/AvroConfluentITCase.java:
##########
@@ -0,0 +1,444 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.schema.registry.test;
+
+import org.apache.flink.connector.testframe.container.FlinkContainers;
+import org.apache.flink.connector.testframe.container.FlinkContainersSettings;
+import org.apache.flink.connector.testframe.container.TestcontainersSettings;
+import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.test.resources.ResourceTestUtils;
+import org.apache.flink.test.util.SQLJobSubmission;
+import org.apache.flink.util.DockerImageVersions;
+import org.apache.flink.util.jackson.JacksonMapperFactory;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.apache.avro.generic.GenericRecord;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.Network;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeoutException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** E2E Test for Avro-Confluent integration. */
+@Testcontainers
+public class AvroConfluentITCase {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AvroConfluentITCase.class);
+
+ private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
+ private static final String INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS = "schema-registry";
+
+ private static final String TOPIC = "test-avro-input";
+ private static final String RESULT_TOPIC = "test-avro-output";
+ private static final String MANUAL_TOPIC = "test-avro-input-manual";
+ private static final String MANUAL_RESULT_TOPIC = "test-avro-output-manual";
+
+ private static final String INPUT_SCHEMA = SchemaLoader.loadSchema("avro/input-record.avsc");
+ private static final String OUTPUT_SCHEMA = SchemaLoader.loadSchema("avro/output-record.avsc");
+
+ private static final Path sqlToolBoxJar = ResourceTestUtils.getResource(".*/SqlToolbox\\.jar");
+
+ private final Path sqlConnectorKafkaJar = ResourceTestUtils.getResource(".*kafka.*\\.jar");
+ private final Path sqlConnectorUpsertTestJar =
+ ResourceTestUtils.getResource(".*flink-test-utils.*\\.jar");
+ private final Path sqlAvroConfluentJar =
+ ResourceTestUtils.getResource(".*avro-confluent.*\\.jar");
+
+ private static final Network NETWORK = Network.newNetwork();
+
+ @Container
+ public static final KafkaContainer KAFKA =
+ new KafkaContainer(DockerImageName.parse(DockerImageVersions.KAFKA))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);
+
+ @Container
+ private static final GenericContainer<?> SCHEMA_REGISTRY =
+ new GenericContainer<>(DockerImageName.parse(DockerImageVersions.SCHEMA_REGISTRY))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS)
+ .withExposedPorts(8081)
+ .withEnv("SCHEMA_REGISTRY_HOST_NAME", INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS)
+ .withEnv(
+ "SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS",
+ INTER_CONTAINER_KAFKA_ALIAS + ":9092")
+ .dependsOn(KAFKA);
+
Review Comment:
You can use
```suggestion
@RegisterExtension
```
to avoid manually starting the Flink Cluster and let Junit handle the lifecycle.
##########
flink-end-to-end-tests/flink-sql-client-test/pom.xml:
##########
@@ -61,14 +61,21 @@ under the License.
<!-- Used by maven-dependency-plugin -->
<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>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-sql-avro-confluent-registry</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
- <version>3.2.3</version>
+ <version>3.9.0</version>
Review Comment:
Nit: SInce you also upgraded the FLink connector to be compatible with Kafka 4.0, can you also use kafka 4.0 here?
##########
flink-end-to-end-tests/flink-confluent-schema-registry/src/test/resources/log4j2-test.properties:
##########
@@ -0,0 +1,34 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+# Set root logger level to OFF to not flood build logs
+# set manually to INFO for debugging purposes
+rootLogger.level = INFO
Review Comment:
Please set this to OFF before merging
##########
flink-end-to-end-tests/flink-confluent-schema-registry/src/test/java/org/apache/flink/schema/registry/test/AvroConfluentITCase.java:
##########
@@ -0,0 +1,444 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.schema.registry.test;
+
+import org.apache.flink.connector.testframe.container.FlinkContainers;
+import org.apache.flink.connector.testframe.container.FlinkContainersSettings;
+import org.apache.flink.connector.testframe.container.TestcontainersSettings;
+import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.test.resources.ResourceTestUtils;
+import org.apache.flink.test.util.SQLJobSubmission;
+import org.apache.flink.util.DockerImageVersions;
+import org.apache.flink.util.jackson.JacksonMapperFactory;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.apache.avro.generic.GenericRecord;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.Network;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeoutException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** E2E Test for Avro-Confluent integration. */
+@Testcontainers
+public class AvroConfluentITCase {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AvroConfluentITCase.class);
+
+ private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
+ private static final String INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS = "schema-registry";
+
+ private static final String TOPIC = "test-avro-input";
+ private static final String RESULT_TOPIC = "test-avro-output";
+ private static final String MANUAL_TOPIC = "test-avro-input-manual";
+ private static final String MANUAL_RESULT_TOPIC = "test-avro-output-manual";
+
+ private static final String INPUT_SCHEMA = SchemaLoader.loadSchema("avro/input-record.avsc");
+ private static final String OUTPUT_SCHEMA = SchemaLoader.loadSchema("avro/output-record.avsc");
+
+ private static final Path sqlToolBoxJar = ResourceTestUtils.getResource(".*/SqlToolbox\\.jar");
+
+ private final Path sqlConnectorKafkaJar = ResourceTestUtils.getResource(".*kafka.*\\.jar");
+ private final Path sqlConnectorUpsertTestJar =
+ ResourceTestUtils.getResource(".*flink-test-utils.*\\.jar");
+ private final Path sqlAvroConfluentJar =
+ ResourceTestUtils.getResource(".*avro-confluent.*\\.jar");
+
+ private static final Network NETWORK = Network.newNetwork();
+
+ @Container
+ public static final KafkaContainer KAFKA =
+ new KafkaContainer(DockerImageName.parse(DockerImageVersions.KAFKA))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);
+
+ @Container
+ private static final GenericContainer<?> SCHEMA_REGISTRY =
+ new GenericContainer<>(DockerImageName.parse(DockerImageVersions.SCHEMA_REGISTRY))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS)
+ .withExposedPorts(8081)
+ .withEnv("SCHEMA_REGISTRY_HOST_NAME", INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS)
+ .withEnv(
+ "SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS",
+ INTER_CONTAINER_KAFKA_ALIAS + ":9092")
+ .dependsOn(KAFKA);
+
+ private static final FlinkContainers FLINK =
+ FlinkContainers.builder()
+ .withFlinkContainersSettings(
+ FlinkContainersSettings.builder().numTaskManagers(1).build())
+ .withTestcontainersSettings(
+ TestcontainersSettings.builder()
+ .network(NETWORK)
+ .logger(LOG)
+ .dependsOn(KAFKA)
+ .build())
+ .build();
+
+ private final HttpClient client = HttpClient.newHttpClient();
+ private final ObjectMapper objectMapper = JacksonMapperFactory.createObjectMapper();
Review Comment:
Nit: I usually make the ObjectMapper static to avoid any reinitialization
##########
docs/content.zh/docs/connectors/table/formats/avro-confluent.md:
##########
@@ -280,6 +280,13 @@ Format 参数
<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 style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether to automatically register schemas with the Confluent Schema Registry if they don't exist. When set to <code>false</code>, schemas must be manually registered in the Schema Registry before being used. When set to <code>true</code>, schemas will be automatically registered during serialization if they don't already exist. The default value is <code>true</code>.</td>
Review Comment:
Did we document anywhere in which scenario we auto register schemas for example when reading or writing from/to a table?
##########
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:
Does it throw an IOException if the schema is not present?
Nit: Can we maybe throw a better exception e.g. FlinkException to make sure this is "expected" on not found schema?
##########
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:
Nit: Can we turn the three (or at least the first/third) tests into a `parameterTest` to avoid the code duplication?
It's a bit unclear to me why only test two verifies the behavior for sinks and sources while the other tests only look at sinks.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "fapaul (via GitHub)" <gi...@apache.org>.
fapaul commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2158395941
##########
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:
Nit: Can we turn the three (or at least the first/third) tests into a `parameterizedTest` to avoid the code duplication?
It's a bit unclear to me why only test two verifies the behavior for sinks and sources while the other tests only look at sinks.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2159125650
##########
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:
Disabling auto registration only means that Flink won't try to register the schema in Schema Registry during every run. However, it still means that the schema that is in Schema Registry, is how Flink would have registered it (so with that specific namespace), or how the user would have provided it via the `avro-confluent.schema` table property.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on PR #26662:
URL: https://github.com/apache/flink/pull/26662#issuecomment-2961350349
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26662:
URL: https://github.com/apache/flink/pull/26662#discussion_r2143064918
##########
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:
I had assumed we would just add here as the optional above is not translated. `<td>yes</td>` as per the note in the link you sent. But it wounds like you have this in hand
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org