You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2020/08/19 11:10:04 UTC
[GitHub] [beam] dennisylyung opened a new pull request #12630: [BEAM-10759] Uses reader Avro schema to deserialize in KafkaIO
dennisylyung opened a new pull request #12630:
URL: https://github.com/apache/beam/pull/12630
The existing ConfluentSchemaRegistryDeserializerProvider uses the writer schema to deserialize Avro records from Kafka, but passes a fixed schema to the AvroCoder. If the Kafka topic contains evolved schema, the AvroCoder will fail to encode the records.
This commit uses the deserialize(bytes, readerSchema) method from KafkaAvroDeserializer, in order to deserialize Kafak Avro records with a schema consistent with AvroCoder.
The change causes ConfluentSchemaRegistryDeserializerProviderTest to fail, due to an existing error. mockRegistryClient.register returns the schema id but not the schema version. With the changes in this commit, the mockRegistryClient will be called one more time, which in turns call mockRegistryClient.register(). This causes schema id and schema version to deviate. Therefore, the unit test is modified to correctly acquire the schema version.
I have ran local test on the java sdk, and spotless test. The modified logic has been tested on an actual pipeline with Dataflow Runner, which successfully read from a topic with evolved schema.
R: @chamikaramj @lukecwik @timrobertson100
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Dataflow | Flink | Samza | Spark | Twister2
--- | --- | --- | --- | --- | --- | ---
Go | [](https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/) | ---
Java | [](https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/)
Python | [](https://ci-beam.apache.org/job/beam_PostCommit_Python2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python35/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python2_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python35_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/) | ---
XLang | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/) | ---
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website
--- | --- | --- | --- | ---
Non-portable | [](https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/) <br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/)
Portable | --- | [](https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | ---
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] iemejia closed pull request #12630: [BEAM-10759] Uses reader Avro schema to deserialize in KafkaIO
Posted by GitBox <gi...@apache.org>.
iemejia closed pull request #12630:
URL: https://github.com/apache/beam/pull/12630
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] iemejia commented on pull request #12630: [BEAM-10759] Uses reader Avro schema to deserialize in KafkaIO
Posted by GitBox <gi...@apache.org>.
iemejia commented on pull request #12630:
URL: https://github.com/apache/beam/pull/12630#issuecomment-698906102
Merged manually to squash the commits.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] dennisylyung commented on pull request #12630: [BEAM-10759] Uses reader Avro schema to deserialize in KafkaIO
Posted by GitBox <gi...@apache.org>.
dennisylyung commented on pull request #12630:
URL: https://github.com/apache/beam/pull/12630#issuecomment-676392473
R: @lukecwik
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] iemejia commented on a change in pull request #12630: [BEAM-10759] Uses reader Avro schema to deserialize in KafkaIO
Posted by GitBox <gi...@apache.org>.
iemejia commented on a change in pull request #12630:
URL: https://github.com/apache/beam/pull/12630#discussion_r474070383
##########
File path: sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ConfluentSchemaRegistryDeserializerProvider.java
##########
@@ -91,7 +91,10 @@
.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl)
.build();
Deserializer<T> deserializer =
- (Deserializer<T>) new KafkaAvroDeserializer(getSchemaRegistryClient());
+ (Deserializer<T>)
+ new ConfluentSchemaRegistryDeserializer(
+ getSchemaRegistryClient(),
+ new Schema.Parser().parse(getSchemaMetadata().getSchema()));
Review comment:
This same line is used in the Coder part for a similar purpose, can we extract maybe this as a private method and reuse?
```
private Schema getAvroSchema() { return new Schema.Parser().parse(getSchemaMetadata().getSchema()); }
``
##########
File path: sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/ConfluentSchemaRegistryDeserializerProviderTest.java
##########
@@ -45,7 +45,8 @@ public void testGetCoder() {
assertEquals(AVRO_SCHEMA, coderV0.getSchema());
try {
- Integer version = mockRegistryClient.register(subject, AVRO_SCHEMA_V1);
+ mockRegistryClient.register(subject, AVRO_SCHEMA_V1);
Review comment:
How is this different? Or better how is this testing that the good schema is chosen? Maybe we can add a specific test for that?
##########
File path: sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ConfluentSchemaRegistryDeserializerProvider.java
##########
@@ -116,3 +119,17 @@ private SchemaRegistryClient getSchemaRegistryClient() {
return this.schemaRegistryClientProviderFn.apply(null);
}
}
+
+class ConfluentSchemaRegistryDeserializer extends KafkaAvroDeserializer {
+ Schema readerSchema;
+
+ ConfluentSchemaRegistryDeserializer(SchemaRegistryClient client, Schema readerSchema) {
+ this.schemaRegistry = client;
Review comment:
super(client);
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] dennisylyung commented on pull request #12630: [BEAM-10759] Uses reader Avro schema to deserialize in KafkaIO
Posted by GitBox <gi...@apache.org>.
dennisylyung commented on pull request #12630:
URL: https://github.com/apache/beam/pull/12630#issuecomment-678366700
Run Java PreCommit
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] dennisylyung commented on pull request #12630: [BEAM-10759] Uses reader Avro schema to deserialize in KafkaIO
Posted by GitBox <gi...@apache.org>.
dennisylyung commented on pull request #12630:
URL: https://github.com/apache/beam/pull/12630#issuecomment-678101761
@iemejia I have pushed a new commit implementing your suggestions.
For the specific test, I edited the mock consumer in KafkaIOTest to generate records with a different schema.
I have verified the new deserializer provider by running the modified test on the original one
```java
@Override
public Deserializer<T> getDeserializer(Map<String, ?> configs, boolean isKey) {
ImmutableMap<String, Object> csrConfig =
ImmutableMap.<String, Object>builder()
.putAll(configs)
.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl)
.build();
Deserializer<T> deserializer =
(Deserializer<T>) new KafkaAvroDeserializer(getSchemaRegistryClient());
// new ConfluentSchemaRegistryDeserializer(getSchemaRegistryClient(), getAvroSchema());
deserializer.configure(csrConfig, isKey);
return deserializer;
}
```
As expected, this will cause the test to fail
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] iemejia commented on a change in pull request #12630: [BEAM-10759] Uses reader Avro schema to deserialize in KafkaIO
Posted by GitBox <gi...@apache.org>.
iemejia commented on a change in pull request #12630:
URL: https://github.com/apache/beam/pull/12630#discussion_r476757936
##########
File path: sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java
##########
@@ -1807,9 +1809,13 @@ void shutdown() {
}
}
+ /**
+ * Mock records with a different schema to test deserializing evolved schema using
Review comment:
It seems this broke the checkstyle validations (First paragraph should finish with period (.). That's the reason why the precommit tests are not passing.
##########
File path: sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java
##########
@@ -1807,9 +1809,13 @@ void shutdown() {
}
}
+ /**
+ * Mock records with a different schema to test deserializing evolved schema using
+ * ConfluentSchemaRegistryDeserializerProvider
+ */
private abstract static class BaseAvroSerializableFunction
implements SerializableFunction<Integer, byte[]> {
- static transient Serializer<AvroGeneratedUser> serializer = null;
+ static transient Serializer<GenericRecord> serializer = null;
Review comment:
We should not change this function because it is used by other tests, maybe we should create a new one that produces GenericRecords and we can parametrize to produce schema specific records.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] iemejia commented on pull request #12630: [BEAM-10759] Uses reader Avro schema to deserialize in KafkaIO
Posted by GitBox <gi...@apache.org>.
iemejia commented on pull request #12630:
URL: https://github.com/apache/beam/pull/12630#issuecomment-698906102
Merged manually to squash the commits.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] iemejia closed pull request #12630: [BEAM-10759] Uses reader Avro schema to deserialize in KafkaIO
Posted by GitBox <gi...@apache.org>.
iemejia closed pull request #12630:
URL: https://github.com/apache/beam/pull/12630
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] dennisylyung commented on pull request #12630: [BEAM-10759] Uses reader Avro schema to deserialize in KafkaIO
Posted by GitBox <gi...@apache.org>.
dennisylyung commented on pull request #12630:
URL: https://github.com/apache/beam/pull/12630#issuecomment-698457493
@iemejia do you have time to review the updates?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] dennisylyung commented on pull request #12630: [BEAM-10759] Uses reader Avro schema to deserialize in KafkaIO
Posted by GitBox <gi...@apache.org>.
dennisylyung commented on pull request #12630:
URL: https://github.com/apache/beam/pull/12630#issuecomment-698457493
@iemejia do you have time to review the updates?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org