You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "klam-shop (via GitHub)" <gi...@apache.org> on 2024/07/10 19:18:51 UTC
[PR] Let ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG overridable by user in `KafkaSourceBuilder` [flink-connector-kafka]
klam-shop opened a new pull request, #108:
URL: https://github.com/apache/flink-connector-kafka/pull/108
Let ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG overridable by user in `KafkaSourceBuilder`, in order to enable the Large Message use-case discussed in this [mailing list discussion](https://lists.apache.org/thread/spl88o63sjm2dv4l5no0ym632d2yt2o6).
This allows users to easily implement the [`claim check` large message pattern](https://developer.confluent.io/patterns/event-processing/claim-check/) without bringing any concerns into the Flink codebase otherwise.
Other Reading:
https://www.kai-waehner.de/blog/2020/08/07/apache-kafka-handling-large-messages-and-files-for-image-video-audio-processing/
https://www.conduktor.io/kafka/how-to-send-large-messages-in-apache-kafka/#Option-1:-using-an-external-store-(GB-size-messages)-0
--
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-35808] Let `ConsumerConfig.(KEY|VALUE)_DESERIALIZER_CLASS_CONFIG` be overridable by user in `KafkaSourceBuilder` [flink-connector-kafka]
Posted by "klam-shop (via GitHub)" <gi...@apache.org>.
klam-shop commented on PR #108:
URL: https://github.com/apache/flink-connector-kafka/pull/108#issuecomment-2223875792
Thanks for your review @fapaul ! I did another pass on trying to incorporate your comments, and also updated some of the documentation that mentions the deserializer configuration.
--
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-35808] Let `ConsumerConfig.(KEY|VALUE)_DESERIALIZER_CLASS_CONFIG` be overridable by user in `KafkaSourceBuilder` [flink-connector-kafka]
Posted by "fapaul (via GitHub)" <gi...@apache.org>.
fapaul commented on PR #108:
URL: https://github.com/apache/flink-connector-kafka/pull/108#issuecomment-2232817971
@klam-shop the CI is failing due to not correclt formatted code https://github.com/apache/flink-connector-kafka/actions/runs/9957480138/job/27552347466?pr=108#step:15:127 can you run `mvn spotless:apply` ?
--
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-35808] Let `ConsumerConfig.(KEY|VALUE)_DESERIALIZER_CLASS_CONFIG` be overridable by user in `KafkaSourceBuilder` [flink-connector-kafka]
Posted by "klam-shop (via GitHub)" <gi...@apache.org>.
klam-shop commented on PR #108:
URL: https://github.com/apache/flink-connector-kafka/pull/108#issuecomment-2233307830
When I ran `mvn spotless:apply` locally, no diffs were made..
```
$ mvn spotless:apply
[INFO] Scanning for projects...
[WARNING]
[WARNING] Some problems were encountered while building the effective model for org.apache.flink:flink-connector-kafka-python:pom:3.3-SNAPSHOT
[WARNING] 'build.plugins.plugin.(groupId:artifactId)' must be unique but found duplicate declaration of plugin org.apache.maven.plugins:maven-dependency-plugin @ line 147, column 21
[WARNING]
[WARNING] It is highly recommended to fix these problems because they threaten the stability of your build.
[WARNING]
[WARNING] For this reason, future Maven versions might no longer support building such malformed projects.
[WARNING]
[INFO] ------------------------------------------------------------------------
[INFO] Reactor Build Order:
[INFO]
[INFO] Flink : Connectors : Kafka : Parent [pom]
[INFO] Flink : Connectors : Kafka [jar]
[INFO] Flink : Connectors : SQL : Kafka [jar]
[INFO] Flink : Connectors : Kafka : E2E Tests [pom]
[INFO] Flink : E2E Tests : Streaming Kafka base [jar]
[INFO] Flink : E2E Tests : Streaming Kafka [jar]
[INFO] Flink : E2E Tests : Common Kafka [jar]
[INFO] Flink : Connectors : SQL : Kafka : Python [pom]
[INFO]
[INFO] -----------< org.apache.flink:flink-connector-kafka-parent >------------
[INFO] Building Flink : Connectors : Kafka : Parent 3.3-SNAPSHOT [1/8]
[INFO] from pom.xml
[INFO] --------------------------------[ pom ]---------------------------------
[INFO]
[INFO] --- spotless:2.27.1:apply (default-cli) @ flink-connector-kafka-parent ---
[INFO] Spotless apply skipped
[INFO]
[INFO] ---------------< org.apache.flink:flink-connector-kafka >---------------
[INFO] Building Flink : Connectors : Kafka 3.3-SNAPSHOT [2/8]
[INFO] from flink-connector-kafka/pom.xml
[INFO] --------------------------------[ jar ]---------------------------------
[INFO]
[INFO] --- spotless:2.27.1:apply (default-cli) @ flink-connector-kafka ---
[INFO] Spotless apply skipped
[INFO]
[INFO] -------------< org.apache.flink:flink-sql-connector-kafka >-------------
[INFO] Building Flink : Connectors : SQL : Kafka 3.3-SNAPSHOT [3/8]
[INFO] from flink-sql-connector-kafka/pom.xml
[INFO] --------------------------------[ jar ]---------------------------------
[INFO]
[INFO] --- spotless:2.27.1:apply (default-cli) @ flink-sql-connector-kafka ---
[INFO] Spotless apply skipped
[INFO]
[INFO] ----------< org.apache.flink:flink-connector-kafka-e2e-tests >----------
[INFO] Building Flink : Connectors : Kafka : E2E Tests 3.3-SNAPSHOT [4/8]
[INFO] from flink-connector-kafka-e2e-tests/pom.xml
[INFO] --------------------------------[ pom ]---------------------------------
[INFO]
[INFO] --- spotless:2.27.1:apply (default-cli) @ flink-connector-kafka-e2e-tests ---
[INFO] Spotless apply skipped
[INFO]
[INFO] ----------< org.apache.flink:flink-streaming-kafka-test-base >----------
[INFO] Building Flink : E2E Tests : Streaming Kafka base 3.3-SNAPSHOT [5/8]
[INFO] from flink-connector-kafka-e2e-tests/flink-streaming-kafka-test-base/pom.xml
[INFO] --------------------------------[ jar ]---------------------------------
[INFO]
[INFO] --- spotless:2.27.1:apply (default-cli) @ flink-streaming-kafka-test-base ---
[INFO] Spotless apply skipped
[INFO]
[INFO] ------------< org.apache.flink:flink-streaming-kafka-test >-------------
[INFO] Building Flink : E2E Tests : Streaming Kafka 3.3-SNAPSHOT [6/8]
[INFO] from flink-connector-kafka-e2e-tests/flink-streaming-kafka-test/pom.xml
[INFO] --------------------------------[ jar ]---------------------------------
[INFO]
[INFO] --- spotless:2.27.1:apply (default-cli) @ flink-streaming-kafka-test ---
[INFO] Spotless apply skipped
[INFO]
[INFO] --------< org.apache.flink:flink-end-to-end-tests-common-kafka >--------
[INFO] Building Flink : E2E Tests : Common Kafka 3.3-SNAPSHOT [7/8]
[INFO] from flink-connector-kafka-e2e-tests/flink-end-to-end-tests-common-kafka/pom.xml
[INFO] --------------------------------[ jar ]---------------------------------
[INFO]
[INFO] --- spotless:2.27.1:apply (default-cli) @ flink-end-to-end-tests-common-kafka ---
[INFO] Spotless apply skipped
[INFO]
[INFO] -----------< org.apache.flink:flink-connector-kafka-python >------------
[INFO] Building Flink : Connectors : SQL : Kafka : Python 3.3-SNAPSHOT [8/8]
[INFO] from flink-python/pom.xml
[INFO] --------------------------------[ pom ]---------------------------------
[INFO]
[INFO] --- spotless:2.27.1:apply (default-cli) @ flink-connector-kafka-python ---
[INFO] Spotless apply skipped
[INFO] ------------------------------------------------------------------------
[INFO] Reactor Summary for Flink : Connectors : Kafka : Parent 3.3-SNAPSHOT:
[INFO]
[INFO] Flink : Connectors : Kafka : Parent ................ SUCCESS [ 0.060 s]
[INFO] Flink : Connectors : Kafka ......................... SUCCESS [ 0.002 s]
[INFO] Flink : Connectors : SQL : Kafka ................... SUCCESS [ 0.001 s]
[INFO] Flink : Connectors : Kafka : E2E Tests ............. SUCCESS [ 0.001 s]
[INFO] Flink : E2E Tests : Streaming Kafka base ........... SUCCESS [ 0.001 s]
[INFO] Flink : E2E Tests : Streaming Kafka ................ SUCCESS [ 0.005 s]
[INFO] Flink : E2E Tests : Common Kafka ................... SUCCESS [ 0.001 s]
[INFO] Flink : Connectors : SQL : Kafka : Python .......... SUCCESS [ 0.001 s]
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 0.188 s
[INFO] Finished at: 2024-07-17T09:18:42-04:00
[INFO] ------------------------------------------------------------------------
```
I tried to apply the fixes locally.
--
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] Let ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG overridable by user in `KafkaSourceBuilder` [flink-connector-kafka]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #108:
URL: https://github.com/apache/flink-connector-kafka/pull/108#issuecomment-2221254845
Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)
--
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-35808] Let `ConsumerConfig.(KEY|VALUE)_DESERIALIZER_CLASS_CONFIG` be overridable by user in `KafkaSourceBuilder` [flink-connector-kafka]
Posted by "fapaul (via GitHub)" <gi...@apache.org>.
fapaul commented on code in PR #108:
URL: https://github.com/apache/flink-connector-kafka/pull/108#discussion_r1677400714
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilderTest.java:
##########
@@ -191,6 +193,72 @@ public void testSettingCustomKafkaSubscriber() {
"Cannot use partitions for consumption because a ExampleCustomSubscriber is already set for consumption.");
}
+ @Test
+ public void testSettingCustomKeyDeserializer() {
+ final String keyDeserializer = TestByteArrayDeserializer.class.getName();
+ final KafkaSource<String> kafkaSource = getBasicBuilder().setProperty("key.deserializer", keyDeserializer).build();
+ // key.deserializer should be overridden
+ assertThat(
+ kafkaSource
+ .getConfiguration()
+ .get(
+ ConfigOptions.key(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG)
+ .stringType()
+ .noDefaultValue()))
+ .isEqualTo(keyDeserializer);
+ }
+
+ @Test
+ public void testSettingCustomValueDeserializer() {
+ final String valueDeserializer = TestByteArrayDeserializer.class.getName();
+ final KafkaSource<String> kafkaSource = getBasicBuilder().setProperty("value.deserializer", valueDeserializer).build();
+ // value.deserializer should be overridden
+ assertThat(
+ kafkaSource
+ .getConfiguration()
+ .get(
+ ConfigOptions.key(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG)
+ .stringType()
+ .noDefaultValue()))
+ .isEqualTo(valueDeserializer);
+ }
+
+ @Test
+ public void testSettingCustomNonByteArrayKeyDeserializer() {
+ final String keyDeserializer = StringDeserializer.class.getName();
+ assertThatThrownBy(() -> getBasicBuilder().setProperty("key.deserializer", keyDeserializer).build())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ String.format("Deserializer class %s does not deserialize byte[]", keyDeserializer));
+ }
+
+ @Test
+ public void testSettingCustomNonByteArrayValueDeserializer() {
+ final String valueDeserializer = StringDeserializer.class.getName();
+ assertThatThrownBy(() -> getBasicBuilder().setProperty("value.deserializer", valueDeserializer).build())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ String.format("Deserializer class %s does not deserialize byte[]", valueDeserializer));
+ }
+
+ @Test
+ public void testSettingCustomNonDeserializer() {
Review Comment:
Can you use parameterized tests here [1]? It should remove the duplicated setup. We can have three test cases and provide the property string as a parameter for those tests.
[1] https://junit.org/junit5/docs/current/user-guide/#writing-tests-parameterized-tests
--
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-35808] Let `ConsumerConfig.(KEY|VALUE)_DESERIALIZER_CLASS_CONFIG` be overridable by user in `KafkaSourceBuilder` [flink-connector-kafka]
Posted by "klam-shop (via GitHub)" <gi...@apache.org>.
klam-shop commented on code in PR #108:
URL: https://github.com/apache/flink-connector-kafka/pull/108#discussion_r1679359612
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilderTest.java:
##########
@@ -191,6 +196,53 @@ public void testSettingCustomKafkaSubscriber() {
"Cannot use partitions for consumption because a ExampleCustomSubscriber is already set for consumption.");
}
+ @Test
+ public void testSettingCustomKeyDeserializer() {
+ final String keyDeserializer = TestByteArrayDeserializer.class.getName();
+ final KafkaSource<String> kafkaSource = getBasicBuilder().setProperty("key.deserializer", keyDeserializer).build();
+ // key.deserializer should be overridden
+ assertThat(
+ kafkaSource
+ .getConfiguration()
+ .get(
+ ConfigOptions.key(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG)
+ .stringType()
+ .noDefaultValue()))
+ .isEqualTo(keyDeserializer);
+ }
+
+ @Test
+ public void testSettingCustomValueDeserializer() {
+ final String valueDeserializer = TestByteArrayDeserializer.class.getName();
+ final KafkaSource<String> kafkaSource = getBasicBuilder().setProperty("value.deserializer", valueDeserializer).build();
+ // value.deserializer should be overridden
+ assertThat(
+ kafkaSource
+ .getConfiguration()
+ .get(
+ ConfigOptions.key(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG)
+ .stringType()
+ .noDefaultValue()))
+ .isEqualTo(valueDeserializer);
+ }
Review Comment:
Ah, my mistake. Fixed that as well
--
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-35808] Let `ConsumerConfig.(KEY|VALUE)_DESERIALIZER_CLASS_CONFIG` be overridable by user in `KafkaSourceBuilder` [flink-connector-kafka]
Posted by "klam-shop (via GitHub)" <gi...@apache.org>.
klam-shop commented on code in PR #108:
URL: https://github.com/apache/flink-connector-kafka/pull/108#discussion_r1679359612
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilderTest.java:
##########
@@ -191,6 +196,53 @@ public void testSettingCustomKafkaSubscriber() {
"Cannot use partitions for consumption because a ExampleCustomSubscriber is already set for consumption.");
}
+ @Test
+ public void testSettingCustomKeyDeserializer() {
+ final String keyDeserializer = TestByteArrayDeserializer.class.getName();
+ final KafkaSource<String> kafkaSource = getBasicBuilder().setProperty("key.deserializer", keyDeserializer).build();
+ // key.deserializer should be overridden
+ assertThat(
+ kafkaSource
+ .getConfiguration()
+ .get(
+ ConfigOptions.key(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG)
+ .stringType()
+ .noDefaultValue()))
+ .isEqualTo(keyDeserializer);
+ }
+
+ @Test
+ public void testSettingCustomValueDeserializer() {
+ final String valueDeserializer = TestByteArrayDeserializer.class.getName();
+ final KafkaSource<String> kafkaSource = getBasicBuilder().setProperty("value.deserializer", valueDeserializer).build();
+ // value.deserializer should be overridden
+ assertThat(
+ kafkaSource
+ .getConfiguration()
+ .get(
+ ConfigOptions.key(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG)
+ .stringType()
+ .noDefaultValue()))
+ .isEqualTo(valueDeserializer);
+ }
Review Comment:
Ah, my mistake. Fixed that as well !
--
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-35808] Let `ConsumerConfig.(KEY|VALUE)_DESERIALIZER_CLASS_CONFIG` be overridable by user in `KafkaSourceBuilder` [flink-connector-kafka]
Posted by "klam-shop (via GitHub)" <gi...@apache.org>.
klam-shop commented on PR #108:
URL: https://github.com/apache/flink-connector-kafka/pull/108#issuecomment-2233615524
>I think I found the culprit. I checked out your branch, and initially, mvn spotless:apply didn't produce any changes, although the check style failed. After switching to Java 11 locally and running mvn spotless:apply, the code formatted correctly.
> I also verified that running mvn org.apache.maven.plugins:maven-checkstyle-plugin:3.1.2:check passes now locally.
Changed to Java 11 locally and ran
```
mvn spotless:apply
mvn org.apache.maven.plugins:maven-checkstyle-plugin:3.1.2:check
```
--
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-35808] Let `ConsumerConfig.(KEY|VALUE)_DESERIALIZER_CLASS_CONFIG` be overridable by user in `KafkaSourceBuilder` [flink-connector-kafka]
Posted by "fapaul (via GitHub)" <gi...@apache.org>.
fapaul merged PR #108:
URL: https://github.com/apache/flink-connector-kafka/pull/108
--
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-35808] Let `ConsumerConfig.(KEY|VALUE)_DESERIALIZER_CLASS_CONFIG` be overridable by user in `KafkaSourceBuilder` [flink-connector-kafka]
Posted by "klam-shop (via GitHub)" <gi...@apache.org>.
klam-shop commented on code in PR #108:
URL: https://github.com/apache/flink-connector-kafka/pull/108#discussion_r1678222384
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilderTest.java:
##########
@@ -191,6 +193,72 @@ public void testSettingCustomKafkaSubscriber() {
"Cannot use partitions for consumption because a ExampleCustomSubscriber is already set for consumption.");
}
+ @Test
+ public void testSettingCustomKeyDeserializer() {
+ final String keyDeserializer = TestByteArrayDeserializer.class.getName();
+ final KafkaSource<String> kafkaSource = getBasicBuilder().setProperty("key.deserializer", keyDeserializer).build();
+ // key.deserializer should be overridden
+ assertThat(
+ kafkaSource
+ .getConfiguration()
+ .get(
+ ConfigOptions.key(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG)
+ .stringType()
+ .noDefaultValue()))
+ .isEqualTo(keyDeserializer);
+ }
+
+ @Test
+ public void testSettingCustomValueDeserializer() {
+ final String valueDeserializer = TestByteArrayDeserializer.class.getName();
+ final KafkaSource<String> kafkaSource = getBasicBuilder().setProperty("value.deserializer", valueDeserializer).build();
+ // value.deserializer should be overridden
+ assertThat(
+ kafkaSource
+ .getConfiguration()
+ .get(
+ ConfigOptions.key(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG)
+ .stringType()
+ .noDefaultValue()))
+ .isEqualTo(valueDeserializer);
+ }
+
+ @Test
+ public void testSettingCustomNonByteArrayKeyDeserializer() {
+ final String keyDeserializer = StringDeserializer.class.getName();
+ assertThatThrownBy(() -> getBasicBuilder().setProperty("key.deserializer", keyDeserializer).build())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ String.format("Deserializer class %s does not deserialize byte[]", keyDeserializer));
+ }
+
+ @Test
+ public void testSettingCustomNonByteArrayValueDeserializer() {
+ final String valueDeserializer = StringDeserializer.class.getName();
+ assertThatThrownBy(() -> getBasicBuilder().setProperty("value.deserializer", valueDeserializer).build())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ String.format("Deserializer class %s does not deserialize byte[]", valueDeserializer));
+ }
+
+ @Test
+ public void testSettingCustomNonDeserializer() {
Review Comment:
Good idea, incorporated this comment.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java:
##########
@@ -548,4 +588,5 @@ private boolean offsetCommitEnabledManually() {
KafkaSourceOptions.COMMIT_OFFSETS_ON_CHECKPOINT.key()));
return autoCommit || commitOnCheckpoint;
}
+
Review Comment:
Thanks, removed this
--
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-35808] Let `ConsumerConfig.(KEY|VALUE)_DESERIALIZER_CLASS_CONFIG` be overridable by user in `KafkaSourceBuilder` [flink-connector-kafka]
Posted by "fapaul (via GitHub)" <gi...@apache.org>.
fapaul commented on code in PR #108:
URL: https://github.com/apache/flink-connector-kafka/pull/108#discussion_r1678856214
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilderTest.java:
##########
@@ -191,6 +196,53 @@ public void testSettingCustomKafkaSubscriber() {
"Cannot use partitions for consumption because a ExampleCustomSubscriber is already set for consumption.");
}
+ @Test
+ public void testSettingCustomKeyDeserializer() {
+ final String keyDeserializer = TestByteArrayDeserializer.class.getName();
+ final KafkaSource<String> kafkaSource = getBasicBuilder().setProperty("key.deserializer", keyDeserializer).build();
+ // key.deserializer should be overridden
+ assertThat(
+ kafkaSource
+ .getConfiguration()
+ .get(
+ ConfigOptions.key(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG)
+ .stringType()
+ .noDefaultValue()))
+ .isEqualTo(keyDeserializer);
+ }
+
+ @Test
+ public void testSettingCustomValueDeserializer() {
+ final String valueDeserializer = TestByteArrayDeserializer.class.getName();
+ final KafkaSource<String> kafkaSource = getBasicBuilder().setProperty("value.deserializer", valueDeserializer).build();
+ // value.deserializer should be overridden
+ assertThat(
+ kafkaSource
+ .getConfiguration()
+ .get(
+ ConfigOptions.key(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG)
+ .stringType()
+ .noDefaultValue()))
+ .isEqualTo(valueDeserializer);
+ }
Review Comment:
Sorry, my initial comment also referred to these tests. The code is identical except the property we set. Can you also convert it to a parameterized test?
--
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-35808] Let `ConsumerConfig.(KEY|VALUE)_DESERIALIZER_CLASS_CONFIG` be overridable by user in `KafkaSourceBuilder` [flink-connector-kafka]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #108:
URL: https://github.com/apache/flink-connector-kafka/pull/108#issuecomment-2236564053
Awesome work, congrats on your first merged pull request!
--
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-35808] Let `ConsumerConfig.(KEY|VALUE)_DESERIALIZER_CLASS_CONFIG` be overridable by user in `KafkaSourceBuilder` [flink-connector-kafka]
Posted by "klam-shop (via GitHub)" <gi...@apache.org>.
klam-shop commented on PR #108:
URL: https://github.com/apache/flink-connector-kafka/pull/108#issuecomment-2229110447
Thanks for the review. Parameterized the tests and ready for another look 🙏
--
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-35808] Let `ConsumerConfig.(KEY|VALUE)_DESERIALIZER_CLASS_CONFIG` be overridable by user in `KafkaSourceBuilder` [flink-connector-kafka]
Posted by "fapaul (via GitHub)" <gi...@apache.org>.
fapaul commented on PR #108:
URL: https://github.com/apache/flink-connector-kafka/pull/108#issuecomment-2233450682
I think I found the culprit. I checked out your branch, and initially, `mvn spotless:apply` didn't produce any changes, although the check style failed. After switching to Java 11 locally and running `mvn spotless:apply,` the code formatted correctly.
I also verified that running `mvn org.apache.maven.plugins:maven-checkstyle-plugin:3.1.2:check` passes now locally.
--
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-35808] Let `ConsumerConfig.(KEY|VALUE)_DESERIALIZER_CLASS_CONFIG` be overridable by user in `KafkaSourceBuilder` [flink-connector-kafka]
Posted by "fapaul (via GitHub)" <gi...@apache.org>.
fapaul commented on code in PR #108:
URL: https://github.com/apache/flink-connector-kafka/pull/108#discussion_r1677401930
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java:
##########
@@ -548,4 +588,5 @@ private boolean offsetCommitEnabledManually() {
KafkaSourceOptions.COMMIT_OFFSETS_ON_CHECKPOINT.key()));
return autoCommit || commitOnCheckpoint;
}
+
Review Comment:
Nit: unrelated change?
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilderTest.java:
##########
@@ -191,6 +193,72 @@ public void testSettingCustomKafkaSubscriber() {
"Cannot use partitions for consumption because a ExampleCustomSubscriber is already set for consumption.");
}
+ @Test
+ public void testSettingCustomKeyDeserializer() {
+ final String keyDeserializer = TestByteArrayDeserializer.class.getName();
+ final KafkaSource<String> kafkaSource = getBasicBuilder().setProperty("key.deserializer", keyDeserializer).build();
+ // key.deserializer should be overridden
+ assertThat(
+ kafkaSource
+ .getConfiguration()
+ .get(
+ ConfigOptions.key(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG)
+ .stringType()
+ .noDefaultValue()))
+ .isEqualTo(keyDeserializer);
+ }
+
+ @Test
+ public void testSettingCustomValueDeserializer() {
+ final String valueDeserializer = TestByteArrayDeserializer.class.getName();
+ final KafkaSource<String> kafkaSource = getBasicBuilder().setProperty("value.deserializer", valueDeserializer).build();
+ // value.deserializer should be overridden
+ assertThat(
+ kafkaSource
+ .getConfiguration()
+ .get(
+ ConfigOptions.key(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG)
+ .stringType()
+ .noDefaultValue()))
+ .isEqualTo(valueDeserializer);
+ }
+
+ @Test
+ public void testSettingCustomNonByteArrayKeyDeserializer() {
+ final String keyDeserializer = StringDeserializer.class.getName();
+ assertThatThrownBy(() -> getBasicBuilder().setProperty("key.deserializer", keyDeserializer).build())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ String.format("Deserializer class %s does not deserialize byte[]", keyDeserializer));
+ }
+
+ @Test
+ public void testSettingCustomNonByteArrayValueDeserializer() {
+ final String valueDeserializer = StringDeserializer.class.getName();
+ assertThatThrownBy(() -> getBasicBuilder().setProperty("value.deserializer", valueDeserializer).build())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ String.format("Deserializer class %s does not deserialize byte[]", valueDeserializer));
+ }
+
+ @Test
+ public void testSettingCustomNonDeserializer() {
Review Comment:
Can you use parameterized tests here [1]? It should remove the duplicated setup. We should have only three test cases and provide the property string as a parameter for those tests.
[1] https://junit.org/junit5/docs/current/user-guide/#writing-tests-parameterized-tests
--
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-35808] Let `ConsumerConfig.(KEY|VALUE)_DESERIALIZER_CLASS_CONFIG` be overridable by user in `KafkaSourceBuilder` [flink-connector-kafka]
Posted by "klam-shop (via GitHub)" <gi...@apache.org>.
klam-shop commented on PR #108:
URL: https://github.com/apache/flink-connector-kafka/pull/108#issuecomment-2233343265
I followed the Code Formatting instructions in https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/flinkdev/ide_setup/#code-formatting and used the IDE to reformat the file.
--
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