You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/01/18 21:23:03 UTC
[GitHub] [flink] JingGe opened a new pull request #18397: [HOTFIX][Kafka] update interface @PublicEvolving or @Internal
JingGe opened a new pull request #18397:
URL: https://github.com/apache/flink/pull/18397
## What is the purpose of the change
The PR added the feasible annotations to some kafka connector interfaces and fixed some code and Javadoc issues.
## Brief change log
- dc25efa3f7
## Verifying this change
This change is a trivial rework / code cleanup and is already covered by existing tests, such as *KafkaSourceBuilderTest* and *KafkaSourceITCase*.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / **no**)
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 89187e8fe1fd38cf50e7f2188cc5cd14c0b65089 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714)
* 2c46519978e62a0f60b8a178d126458d3c11da4b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912)
* 386e9d632753391d80bbbcd41dacb6a1786d5727 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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r788957396
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaRecordDeserializationSchema.java
##########
@@ -123,9 +121,8 @@ default void open(DeserializationSchema.InitializationContext context) throws Ex
* @return A {@link KafkaRecordDeserializationSchema} that deserialize the value with the given
* deserializer.
*/
- static <V, D extends Configurable & Deserializer<V>>
- KafkaRecordDeserializationSchema<V> valueOnly(
- Class<D> valueDeserializerClass, Map<String, String> config) {
+ static <V, D extends Deserializer<V>> KafkaRecordDeserializationSchema<V> valueOnly(
Review comment:
Thanks for the review.
Backward compatibility has been considered. This PR just reduced the type control of the input parameter DeserializationSchema a little bit from `Configurable & Deserializer<V>` to `Deserializer<V>`. The `[Wrapper](https://github.com/apache/flink/pull/18397/files#diff-ed43df5179d747c55bc2b9478ead0b15d49a5275e76827121d31904db779449fL68)` will check internally and call `org.apache.kafka.common.Configurable.configure` if the schema implements `Configurable`.
--
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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r788957396
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaRecordDeserializationSchema.java
##########
@@ -123,9 +121,8 @@ default void open(DeserializationSchema.InitializationContext context) throws Ex
* @return A {@link KafkaRecordDeserializationSchema} that deserialize the value with the given
* deserializer.
*/
- static <V, D extends Configurable & Deserializer<V>>
- KafkaRecordDeserializationSchema<V> valueOnly(
- Class<D> valueDeserializerClass, Map<String, String> config) {
+ static <V, D extends Deserializer<V>> KafkaRecordDeserializationSchema<V> valueOnly(
Review comment:
Thanks for the review.
Backward compatibility has been considered. This PR just reduced the type control of the input parameter DeserializationSchema a little bit from `Configurable & Deserializer<V>` to `Deserializer<V>`. The [Wrapper](https://github.com/apache/flink/pull/18397/files#diff-ed43df5179d747c55bc2b9478ead0b15d49a5275e76827121d31904db779449fL68) will check internally and call `org.apache.kafka.common.Configurable.configure` if the schema implements `Configurable`.
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
}, {
"hash" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921",
"triggerID" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 89187e8fe1fd38cf50e7f2188cc5cd14c0b65089 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714)
* 2c46519978e62a0f60b8a178d126458d3c11da4b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912)
* 386e9d632753391d80bbbcd41dacb6a1786d5727 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916)
* 5b2d44dfea90aa64e7f9c2e65307604abb5ad556 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921)
<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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
}, {
"hash" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921",
"triggerID" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"triggerType" : "PUSH"
}, {
"hash" : "3e1198a17fea03035bc621268a39ea3b386809a4",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30030",
"triggerID" : "3e1198a17fea03035bc621268a39ea3b386809a4",
"triggerType" : "PUSH"
}, {
"hash" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30080",
"triggerID" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3e1198a17fea03035bc621268a39ea3b386809a4 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30030)
* 1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30080)
<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
[GitHub] [flink] lindong28 commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r791270225
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilderTest.java
##########
@@ -160,6 +165,37 @@ public void testSerializeRecordWithKey() {
assertArrayEquals(record.value(), serializationSchema.serialize("a"));
}
+ @Test
+ public void testNoConfigurableKafkaKeySerializer() throws Exception {
+ final Map<String, String> config = ImmutableMap.of("simpleKey", "simpleValue");
+ final KafkaRecordSerializationSchema<String> schema =
+ KafkaRecordSerializationSchema.builder()
+ .setTopic(DEFAULT_TOPIC)
+ // use StringSerializer as dummy Serializer, since ValueSerializer is
Review comment:
Hmm.. It looks like all Javadoc (e.g. method, API) starts with first letter being capital. But many non-Javadoc comments (e.g. comments in the function body) starts with first letter being lower case. You are right that this seems inconsistent...
I didn't find a rule for this on https://flink.apache.org/contributing/code-style-and-quality-common.html. I was primarily following Javadoc code style which almost always starts with an upper-case letter.
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -36,10 +36,15 @@
/** A package private class to wrap {@link Deserializer}. */
class KafkaValueOnlyDeserializerWrapper<T> implements KafkaRecordDeserializationSchema<T> {
+
private static final long serialVersionUID = 5409547407386004054L;
+
private static final Logger LOG =
LoggerFactory.getLogger(KafkaValueOnlyDeserializerWrapper.class);
+
private final Class<? extends Deserializer<T>> deserializerClass;
+ // always be false since this Deserializer is only used for value.
Review comment:
Thank you Jing for digging into 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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790939938
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -36,10 +36,15 @@
/** A package private class to wrap {@link Deserializer}. */
class KafkaValueOnlyDeserializerWrapper<T> implements KafkaRecordDeserializationSchema<T> {
+
private static final long serialVersionUID = 5409547407386004054L;
+
private static final Logger LOG =
LoggerFactory.getLogger(KafkaValueOnlyDeserializerWrapper.class);
+
private final Class<? extends Deserializer<T>> deserializerClass;
+ // always be false since this Deserializer is only used for value.
Review comment:
Code style consistency means in this case that developers, when they come from `KafkaSerializerWrapper` will check the `isKey` at first to see if `deserializer.configure(...)` has been used in `KafkaValueOnlyDeserializerWrapper`. It is much easier to check it via instance variable than digging into the code. If you would try to read both classes back and forth, you would get the feeling.
It is not a big deal. I will change it back.
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
}, {
"hash" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921",
"triggerID" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"triggerType" : "PUSH"
}, {
"hash" : "3e1198a17fea03035bc621268a39ea3b386809a4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30030",
"triggerID" : "3e1198a17fea03035bc621268a39ea3b386809a4",
"triggerType" : "PUSH"
}, {
"hash" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30080",
"triggerID" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"triggerType" : "PUSH"
}, {
"hash" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30080",
"triggerID" : "1020716290",
"triggerType" : "MANUAL"
}, {
"hash" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30080",
"triggerID" : "1020958323",
"triggerType" : "MANUAL"
} ]
}-->
## CI report:
* 1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30080)
<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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 89187e8fe1fd38cf50e7f2188cc5cd14c0b65089 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714)
<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
[GitHub] [flink] fapaul commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
fapaul commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r788958657
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaRecordDeserializationSchema.java
##########
@@ -123,9 +121,8 @@ default void open(DeserializationSchema.InitializationContext context) throws Ex
* @return A {@link KafkaRecordDeserializationSchema} that deserialize the value with the given
* deserializer.
*/
- static <V, D extends Configurable & Deserializer<V>>
- KafkaRecordDeserializationSchema<V> valueOnly(
- Class<D> valueDeserializerClass, Map<String, String> config) {
+ static <V, D extends Deserializer<V>> KafkaRecordDeserializationSchema<V> valueOnly(
Review comment:
@lindong28 thanks for your feedback. I am wondering a bit what the best way here is. Removing the method seems not ideal because as you said users might have used it with Serializers/Deserializers implementing Configurable.
In the Kafka ecosystem there seem to be different ways of configuring a Serializer/Deserializer one is the current one with implementing Configurable and the other is through the Serializer/Deserializer itself [1]. I am no Kafka expert so I don't know the right way to configure them. I guess we can also support both methods.
[1] https://github.com/apache/kafka/blob/3dd3f3ef035501816efac23c32d517bf62fdaa56/clients/src/main/java/org/apache/kafka/common/serialization/Serializer.java#L40
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
}, {
"hash" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921",
"triggerID" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2c46519978e62a0f60b8a178d126458d3c11da4b Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912)
* 386e9d632753391d80bbbcd41dacb6a1786d5727 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916)
* 5b2d44dfea90aa64e7f9c2e65307604abb5ad556 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921)
<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
[GitHub] [flink] lindong28 commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790076382
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSerializerWrapper.java
##########
@@ -32,26 +32,46 @@
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;
+/** Wrapper for Kafka {@link Serializer}. */
class KafkaSerializerWrapper<IN> implements SerializationSchema<IN> {
+
private final Class<? extends Serializer<? super IN>> serializerClass;
+ // whether the serializer is for key or value
+ private final boolean isKey;
private final Map<String, String> config;
private final Function<? super IN, String> topicSelector;
private transient Serializer<? super IN> serializer;
KafkaSerializerWrapper(
Class<? extends Serializer<? super IN>> serializerClass,
+ boolean isKey,
Map<String, String> config,
Function<? super IN, String> topicSelector) {
this.serializerClass = checkNotNull(serializerClass);
+ this.isKey = isKey;
this.config = checkNotNull(config);
this.topicSelector = checkNotNull(topicSelector);
}
KafkaSerializerWrapper(
Class<? extends Serializer<? super IN>> serializerClass,
+ Map<String, String> config,
+ Function<? super IN, String> topicSelector) {
+ this(serializerClass, false, config, topicSelector);
+ }
+
+ KafkaSerializerWrapper(
Review comment:
nits: Previously we have two constructors here. Now we have 4 constructors due to the addition of `isKey` parameter.
Would it be simpler to keep only two constructors and let caller always explicitly provides the `isKey` value? This probably also makes the `KafkaRecordSerializationSchemaBuilder` code more readable.
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -36,10 +36,15 @@
/** A package private class to wrap {@link Deserializer}. */
class KafkaValueOnlyDeserializerWrapper<T> implements KafkaRecordDeserializationSchema<T> {
+
private static final long serialVersionUID = 5409547407386004054L;
+
private static final Logger LOG =
LoggerFactory.getLogger(KafkaValueOnlyDeserializerWrapper.class);
+
private final Class<? extends Deserializer<T>> deserializerClass;
+ // always be false since this Deserializer is only used for value.
Review comment:
nits: The comment seems to be outdated since this value could be true.
And could we start the comment with an upper-case letter for consistency with other comments?
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilderTest.java
##########
@@ -53,11 +54,15 @@
private static final String DEFAULT_TOPIC = "test";
+ private static Map<String, ?> configurableConfiguration;
Review comment:
hmm... would it be simpler to let `ConfigurableStringSerializer` use the `configuration` below instead of creating this `configurableConfiguration`?
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaRecordDeserializationSchema.java
##########
@@ -108,24 +107,24 @@ default void open(DeserializationSchema.InitializationContext context) throws Ex
*/
static <V> KafkaRecordDeserializationSchema<V> valueOnly(
Class<? extends Deserializer<V>> valueDeserializerClass) {
- return new KafkaValueOnlyDeserializerWrapper<>(
- valueDeserializerClass, Collections.emptyMap());
+ return valueOnly(valueDeserializerClass, Collections.emptyMap());
}
/**
* Wraps a Kafka {@link Deserializer} to a {@link KafkaRecordDeserializationSchema}.
*
* @param valueDeserializerClass the deserializer class used to deserialize the value.
- * @param config the configuration of the value deserializer, only valid when the deserializer
- * is an implementation of {@code Configurable}.
+ * @param config the configuration of the value deserializer. If the deserializer is an
Review comment:
nits: the doc seems to miss the explanation of what is invoked when the deserializer is NOT an implementation of Configurable.
Would it be slightly better to use something like this:
```
The configuration of the value deserializer. If the deserializer is an
implementation of {@code Configurable},
{@link org.apache.kafka.common.Configurable#configure(Map)}
will be invoked with the given config. Otherwise, {@link Deserializer#configure(Map, boolean)}
will be invoked with the given config.
```
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 89187e8fe1fd38cf50e7f2188cc5cd14c0b65089 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714)
* 2c46519978e62a0f60b8a178d126458d3c11da4b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912)
<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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
}, {
"hash" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 89187e8fe1fd38cf50e7f2188cc5cd14c0b65089 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714)
* 2c46519978e62a0f60b8a178d126458d3c11da4b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912)
* 386e9d632753391d80bbbcd41dacb6a1786d5727 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916)
* 5b2d44dfea90aa64e7f9c2e65307604abb5ad556 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
[GitHub] [flink] lindong28 commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790751091
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilderTest.java
##########
@@ -53,11 +54,15 @@
private static final String DEFAULT_TOPIC = "test";
+ private static Map<String, ?> configurableConfiguration;
Review comment:
You are right. Thanks for the explanation.
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
}, {
"hash" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921",
"triggerID" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"triggerType" : "PUSH"
}, {
"hash" : "3e1198a17fea03035bc621268a39ea3b386809a4",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30030",
"triggerID" : "3e1198a17fea03035bc621268a39ea3b386809a4",
"triggerType" : "PUSH"
}, {
"hash" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30080",
"triggerID" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"triggerType" : "PUSH"
}, {
"hash" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30080",
"triggerID" : "1020716290",
"triggerType" : "MANUAL"
} ]
}-->
## CI report:
* 3e1198a17fea03035bc621268a39ea3b386809a4 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30030)
* 1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30080)
<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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dc25efa3f7f3488c227b79c99d5639037ed35655 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664)
* 89187e8fe1fd38cf50e7f2188cc5cd14c0b65089 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714)
<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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r789602784
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSerializerWrapper.java
##########
@@ -65,8 +77,15 @@ public void open(InitializationContext context) throws Exception {
serializerClass.getName(),
Serializer.class,
getClass().getClassLoader());
+
+ if (config.isEmpty()) {
Review comment:
I have thought about it. The question is why should user call `configure(...)` with an empty config?
Allowing empty config going through sounds like a backdoor that the Map reference is injected into `configure(...)`. User might want to change the content of the map outside of the serializer/deserializer and have effect on exist serializer/deserializer retroactively(hacking, which is not a good idea), if the implementation of serializer/deserializer is not done correctly. The config map as input should be used as a one-off source and should be even turned into an immutable map before hand it over to the `Configurable/Deserializer::configure(...)` call.
User will be aware that nothing will be done with an empty config where e.g. content were forgot to add by mistake. How about adding logger so that any potential issues will be trackable in the log files.
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 89187e8fe1fd38cf50e7f2188cc5cd14c0b65089 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714)
* 2c46519978e62a0f60b8a178d126458d3c11da4b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912)
* 386e9d632753391d80bbbcd41dacb6a1786d5727 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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r789726411
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSerializerWrapper.java
##########
@@ -65,8 +77,15 @@ public void open(InitializationContext context) throws Exception {
serializerClass.getName(),
Serializer.class,
getClass().getClassLoader());
+
+ if (config.isEmpty()) {
Review comment:
fair enough, I will remove this part.
--
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
[GitHub] [flink] fapaul commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
fapaul commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r789453203
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -62,8 +75,15 @@ public void open(DeserializationSchema.InitializationContext context) throws Exc
deserializerClass.getName(),
Deserializer.class,
getClass().getClassLoader());
+
+ if (config.isEmpty()) {
+ return;
+ }
+
if (deserializer instanceof Configurable) {
((Configurable) deserializer).configure(config);
+ } else {
Review comment:
Currently, we have the following scenarios:
1. `De/Serializer implements Configurable` I would **only** call `Configurable.configure`
2. Does not implement Configurable but a configuration is given call `De/Serializer.configure`
3. No configuration is given we call `De/Serializer.configure` with an empty map?
Probably you have to write tests for all of them although I am not fully sure about the last. I guess calling `configure` with an empty map does not hurt.
--
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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790600948
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilderTest.java
##########
@@ -53,11 +54,15 @@
private static final String DEFAULT_TOPIC = "test";
+ private static Map<String, ?> configurableConfiguration;
Review comment:
I understood your concern. I've considered this option and chose the current solution with the following reasons:
- the business cases `ConfigurableStringSerializer` and `SimpleStringSerializer` simulated are isolated, i.e. there is no way to used them simultaneously. The test env should stick to the real business case as more as possible, therefore shared map should not be used to mix them up.
- test code will be maintained for long time and by many different developers. It has higher priority to make the test code straightforward, i.e. KISS and being able to avoid/reduce human mistakes as far as possible. By default, the cost of using a new HashMap is about 144 bytes (could be reduced to be less), i.e. 16 empty buckets when the hashMap has been created plus loadFactor, modCount, threshold, size, and header. The benefit is a much clean and safe way to write the test. The benefit is large enough with the minimal cost. I think the ROI is great.
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
}, {
"hash" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921",
"triggerID" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"triggerType" : "PUSH"
}, {
"hash" : "3e1198a17fea03035bc621268a39ea3b386809a4",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3e1198a17fea03035bc621268a39ea3b386809a4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5b2d44dfea90aa64e7f9c2e65307604abb5ad556 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921)
* 3e1198a17fea03035bc621268a39ea3b386809a4 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
[GitHub] [flink] lindong28 commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790792173
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -36,10 +36,15 @@
/** A package private class to wrap {@link Deserializer}. */
class KafkaValueOnlyDeserializerWrapper<T> implements KafkaRecordDeserializationSchema<T> {
+
private static final long serialVersionUID = 5409547407386004054L;
+
private static final Logger LOG =
LoggerFactory.getLogger(KafkaValueOnlyDeserializerWrapper.class);
+
private final Class<? extends Deserializer<T>> deserializerClass;
+ // always be false since this Deserializer is only used for value.
Review comment:
Hmm... do you mean we add `isKey` here to be consistent with `KafkaSerializerWrapper`? I am not sure we need to do this. Note that `KafkaSerializerWrapper`'s constructor takes `isKey` as parameter but `KafkaValueOnlyDeserializerWrapper` does not. So these two classes will anyway be inconsistent.
Maybe @fapaul can also comment on 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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790964213
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilderTest.java
##########
@@ -160,6 +165,37 @@ public void testSerializeRecordWithKey() {
assertArrayEquals(record.value(), serializationSchema.serialize("a"));
}
+ @Test
+ public void testNoConfigurableKafkaKeySerializer() throws Exception {
+ final Map<String, String> config = ImmutableMap.of("simpleKey", "simpleValue");
+ final KafkaRecordSerializationSchema<String> schema =
+ KafkaRecordSerializationSchema.builder()
+ .setTopic(DEFAULT_TOPIC)
+ // use StringSerializer as dummy Serializer, since ValueSerializer is
Review comment:
I'll change it. Just out of curiosity, I didn't find the consistency rule of single line comment with the first letter being capital. There are also many opposite examples in the codebase, e.g. https://github.com/JingGe/flink/blob/master/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/StreamFormatAdapter.java
https://github.com/JingGe/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilderTest.java
##########
@@ -160,6 +165,37 @@ public void testSerializeRecordWithKey() {
assertArrayEquals(record.value(), serializationSchema.serialize("a"));
}
+ @Test
+ public void testNoConfigurableKafkaKeySerializer() throws Exception {
+ final Map<String, String> config = ImmutableMap.of("simpleKey", "simpleValue");
+ final KafkaRecordSerializationSchema<String> schema =
+ KafkaRecordSerializationSchema.builder()
+ .setTopic(DEFAULT_TOPIC)
+ // use StringSerializer as dummy Serializer, since ValueSerializer is
Review comment:
I'll change it. Just out of curiosity, I didn't find the consistency rule of single line comment with the first letter being capital. There are also many opposite examples using lowercase in the codebase, e.g. https://github.com/JingGe/flink/blob/master/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/StreamFormatAdapter.java
https://github.com/JingGe/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilderTest.java
##########
@@ -160,6 +165,37 @@ public void testSerializeRecordWithKey() {
assertArrayEquals(record.value(), serializationSchema.serialize("a"));
}
+ @Test
+ public void testNoConfigurableKafkaKeySerializer() throws Exception {
+ final Map<String, String> config = ImmutableMap.of("simpleKey", "simpleValue");
+ final KafkaRecordSerializationSchema<String> schema =
+ KafkaRecordSerializationSchema.builder()
+ .setTopic(DEFAULT_TOPIC)
+ // use StringSerializer as dummy Serializer, since ValueSerializer is
Review comment:
I'll change it. Just out of curiosity, I didn't find the consistency rule of single line comment with the first letter being capital. There are also many opposite examples using lowercase in the codebase, e.g. https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/StreamFormatAdapter.java
https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -36,10 +36,15 @@
/** A package private class to wrap {@link Deserializer}. */
class KafkaValueOnlyDeserializerWrapper<T> implements KafkaRecordDeserializationSchema<T> {
+
private static final long serialVersionUID = 5409547407386004054L;
+
private static final Logger LOG =
LoggerFactory.getLogger(KafkaValueOnlyDeserializerWrapper.class);
+
private final Class<? extends Deserializer<T>> deserializerClass;
+ // always be false since this Deserializer is only used for value.
Review comment:
Code style consistency means in this case that developers, when they come from `KafkaSerializerWrapper` will check the `isKey` at first to see if `deserializer.configure(...)` has been used in `KafkaValueOnlyDeserializerWrapper`. It is much easier to check it via instance variable than digging into the code. If you would try to read both classes back and forth, you would get the feeling.
It is not a big deal. I will change it back.
--
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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790964213
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilderTest.java
##########
@@ -160,6 +165,37 @@ public void testSerializeRecordWithKey() {
assertArrayEquals(record.value(), serializationSchema.serialize("a"));
}
+ @Test
+ public void testNoConfigurableKafkaKeySerializer() throws Exception {
+ final Map<String, String> config = ImmutableMap.of("simpleKey", "simpleValue");
+ final KafkaRecordSerializationSchema<String> schema =
+ KafkaRecordSerializationSchema.builder()
+ .setTopic(DEFAULT_TOPIC)
+ // use StringSerializer as dummy Serializer, since ValueSerializer is
Review comment:
I'll change it. Just out of curiosity, I didn't find the consistency rule of single line comment with the first letter being capital. There are also many opposite examples in the codebase, e.g. https://github.com/JingGe/flink/blob/master/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/StreamFormatAdapter.java
--
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
[GitHub] [flink] lindong28 commented on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
lindong28 commented on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1020716213
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 89187e8fe1fd38cf50e7f2188cc5cd14c0b65089 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714)
* 2c46519978e62a0f60b8a178d126458d3c11da4b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912)
* 386e9d632753391d80bbbcd41dacb6a1786d5727 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916)
<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
[GitHub] [flink] lindong28 commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r789704887
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSerializerWrapper.java
##########
@@ -65,8 +77,15 @@ public void open(InitializationContext context) throws Exception {
serializerClass.getName(),
Serializer.class,
getClass().getClassLoader());
+
+ if (config.isEmpty()) {
Review comment:
It is possible that relying on an invocation of `configure(...)` with an empty config is a bad practice. However, since this is an existing API that public users might already be using, users will be unhappy if our change breaks their existing application, right? I am just not sure it is a good idea to tell users that we break their application because we believe their usage is wrong.
Also, the public method will probably be easier to understand/use if its semantics is `deserializer::configure(...) will be invoked with the config map provided`, rather than saying `deserializer::configure(...) will be invoked with the config map provided if the map is empty`. What do you think?
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 89187e8fe1fd38cf50e7f2188cc5cd14c0b65089 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714)
* 2c46519978e62a0f60b8a178d126458d3c11da4b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912)
<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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
}, {
"hash" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921",
"triggerID" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"triggerType" : "PUSH"
}, {
"hash" : "3e1198a17fea03035bc621268a39ea3b386809a4",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30030",
"triggerID" : "3e1198a17fea03035bc621268a39ea3b386809a4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5b2d44dfea90aa64e7f9c2e65307604abb5ad556 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921)
* 3e1198a17fea03035bc621268a39ea3b386809a4 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30030)
<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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r788965796
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -62,8 +75,15 @@ public void open(DeserializationSchema.InitializationContext context) throws Exc
deserializerClass.getName(),
Deserializer.class,
getClass().getClassLoader());
+
+ if (config.isEmpty()) {
+ return;
+ }
+
if (deserializer instanceof Configurable) {
((Configurable) deserializer).configure(config);
+ } else {
Review comment:
Do you mean if it didn't implement Configurable, then `Deserializer.configure` will be called?
--
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
[GitHub] [flink] fapaul commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
fapaul commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790810201
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -36,10 +36,15 @@
/** A package private class to wrap {@link Deserializer}. */
class KafkaValueOnlyDeserializerWrapper<T> implements KafkaRecordDeserializationSchema<T> {
+
private static final long serialVersionUID = 5409547407386004054L;
+
private static final Logger LOG =
LoggerFactory.getLogger(KafkaValueOnlyDeserializerWrapper.class);
+
private final Class<? extends Deserializer<T>> deserializerClass;
+ // always be false since this Deserializer is only used for value.
Review comment:
Hmm, I'd favor removing the instance variable and directly calling `deserializer.configure(config, false);` and move the comment above this line
--
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
[GitHub] [flink] flinkbot commented on pull request #18397: [HOTFIX][Kafka] update interface @PublicEvolving or @Internal
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848712
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit dc25efa3f7f3488c227b79c99d5639037ed35655 (Tue Jan 18 21:27:20 UTC 2022)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
* **Invalid pull request title: No valid Jira ID provided**
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r788957396
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaRecordDeserializationSchema.java
##########
@@ -123,9 +121,8 @@ default void open(DeserializationSchema.InitializationContext context) throws Ex
* @return A {@link KafkaRecordDeserializationSchema} that deserialize the value with the given
* deserializer.
*/
- static <V, D extends Configurable & Deserializer<V>>
- KafkaRecordDeserializationSchema<V> valueOnly(
- Class<D> valueDeserializerClass, Map<String, String> config) {
+ static <V, D extends Deserializer<V>> KafkaRecordDeserializationSchema<V> valueOnly(
Review comment:
Thanks for the review.
Backward compatibility has been considered. This PR just reduced the type control of the input parameter DeserializationSchema a little bit from `Configurable & Deserializer<V>` to `Deserializer<V>`. The [Wrapper](https://github.com/apache/flink/pull/18397/files#diff-ed43df5179d747c55bc2b9478ead0b15d49a5275e76827121d31904db779449fL68) will check internally and call `org.apache.kafka.common.Configurable.configure` if the schema implements `Configurable`. In this way, old cases due to the backward compatibility will be supported implicitly without confusing users who use `Deserializer.configure`
--
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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r788957396
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaRecordDeserializationSchema.java
##########
@@ -123,9 +121,8 @@ default void open(DeserializationSchema.InitializationContext context) throws Ex
* @return A {@link KafkaRecordDeserializationSchema} that deserialize the value with the given
* deserializer.
*/
- static <V, D extends Configurable & Deserializer<V>>
- KafkaRecordDeserializationSchema<V> valueOnly(
- Class<D> valueDeserializerClass, Map<String, String> config) {
+ static <V, D extends Deserializer<V>> KafkaRecordDeserializationSchema<V> valueOnly(
Review comment:
Thanks for the review.
Backward compatibility has been considered. This PR just reduced the type control of the input parameter DeserializationSchema a little bit from `Configurable & Deserializer<V>` to `Deserializer<V>`. The [Wrapper](https://github.com/apache/flink/pull/18397/files#diff-ed43df5179d747c55bc2b9478ead0b15d49a5275e76827121d31904db779449fL68) will check internally and call `org.apache.kafka.common.Configurable.configure` if the schema implements `Configurable`. In this way, older case will be supported implicitly without confusing users who use `Deserializer.configure`
--
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
[GitHub] [flink] JingGe commented on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1020958323
@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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
}, {
"hash" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921",
"triggerID" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"triggerType" : "PUSH"
}, {
"hash" : "3e1198a17fea03035bc621268a39ea3b386809a4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30030",
"triggerID" : "3e1198a17fea03035bc621268a39ea3b386809a4",
"triggerType" : "PUSH"
}, {
"hash" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30080",
"triggerID" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"triggerType" : "PUSH"
}, {
"hash" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30080",
"triggerID" : "1020716290",
"triggerType" : "MANUAL"
} ]
}-->
## CI report:
* 1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30080)
<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
[GitHub] [flink] lindong28 commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790786440
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaRecordDeserializationSchemaTest.java
##########
@@ -93,6 +112,29 @@ public void testKafkaValueDeserializerWrapper() throws Exception {
assertEquals("world", collector.list.get(0));
}
+ @Test
+ public void testNoConfigurableKafkaSerializer() throws Exception {
+ final Map<String, String> config = ImmutableMap.of("simpleKey", "simpleValue");
+ KafkaRecordDeserializationSchema<String> schema =
+ KafkaRecordDeserializationSchema.valueOnly(SimpleStringSerializer.class, config);
+ schema.open(new TestingDeserializationContext());
+ Assertions.assertEquals(configuration, config);
+ assertFalse(isKeyDeserializer);
+ assertTrue(configurableConfiguration.isEmpty());
+ }
+
+ @Test
+ public void testConfigurableKafkaValueSerializer() throws Exception {
Review comment:
nits: The test name should probably mention `deserializer` instead of `serializer`.
How about naming it `testKafkaValueDeserializerWrapperWithConfigurable` so that its name is more consistent with the existing `testKafkaValueDeserializerWrapper`?
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSerializerWrapper.java
##########
@@ -32,26 +32,33 @@
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;
+/** Wrapper for Kafka {@link Serializer}. */
class KafkaSerializerWrapper<IN> implements SerializationSchema<IN> {
+
private final Class<? extends Serializer<? super IN>> serializerClass;
+ // whether the serializer is for key or value
Review comment:
nits: could we change `whether` to `Whether`?
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaRecordDeserializationSchemaTest.java
##########
@@ -93,6 +112,29 @@ public void testKafkaValueDeserializerWrapper() throws Exception {
assertEquals("world", collector.list.get(0));
}
+ @Test
+ public void testNoConfigurableKafkaSerializer() throws Exception {
Review comment:
nits: The test name should probably mention `deserializer` instead of `serializer`.
How about naming it `testKafkaValueDeserializerWrapperWithoutConfigurable` so that its name is more consistent with the existing `testKafkaValueDeserializerWrapper`?
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilderTest.java
##########
@@ -160,6 +165,37 @@ public void testSerializeRecordWithKey() {
assertArrayEquals(record.value(), serializationSchema.serialize("a"));
}
+ @Test
+ public void testNoConfigurableKafkaKeySerializer() throws Exception {
+ final Map<String, String> config = ImmutableMap.of("simpleKey", "simpleValue");
+ final KafkaRecordSerializationSchema<String> schema =
+ KafkaRecordSerializationSchema.builder()
+ .setTopic(DEFAULT_TOPIC)
+ // use StringSerializer as dummy Serializer, since ValueSerializer is
Review comment:
nits: `use` -> `Use`
--
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
[GitHub] [flink] lindong28 commented on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
lindong28 commented on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1020716213
--
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
[GitHub] [flink] lindong28 commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r791270225
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilderTest.java
##########
@@ -160,6 +165,37 @@ public void testSerializeRecordWithKey() {
assertArrayEquals(record.value(), serializationSchema.serialize("a"));
}
+ @Test
+ public void testNoConfigurableKafkaKeySerializer() throws Exception {
+ final Map<String, String> config = ImmutableMap.of("simpleKey", "simpleValue");
+ final KafkaRecordSerializationSchema<String> schema =
+ KafkaRecordSerializationSchema.builder()
+ .setTopic(DEFAULT_TOPIC)
+ // use StringSerializer as dummy Serializer, since ValueSerializer is
Review comment:
Hmm.. It looks like all Javadoc (e.g. method, API) starts with first letter being capital. But many non-Javadoc comments (e.g. comments in the function body) starts with first letter being lower case. You are right that this seems inconsistent...
I didn't find a rule for this on https://flink.apache.org/contributing/code-style-and-quality-common.html. I was primarily following Javadoc code style which almost always starts with an upper-case letter.
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [HOTFIX][Kafka] update interface @PublicEvolving or @Internal
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dc25efa3f7f3488c227b79c99d5639037ed35655 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664)
<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
[GitHub] [flink] lindong28 commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r789704887
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSerializerWrapper.java
##########
@@ -65,8 +77,15 @@ public void open(InitializationContext context) throws Exception {
serializerClass.getName(),
Serializer.class,
getClass().getClassLoader());
+
+ if (config.isEmpty()) {
Review comment:
It is possible that relying on an invocation of `configure(...)` with an empty config is a bad practice. However, since this is an existing API that public users might already be using, users will be unhappy if our change break their existing application, right? I am just not sure it is a good idea to tell users that we break their application because we believe their usage is wrong.
Also, the public method will probably be easier to understand/use if its semantics is `deserializer::configure(...) will be invoked with the config map provided`, rather than saying `deserializer::configure(...) will be invoked with the config map provided if the map is empty`. What do you think?
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
}, {
"hash" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921",
"triggerID" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 386e9d632753391d80bbbcd41dacb6a1786d5727 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916)
* 5b2d44dfea90aa64e7f9c2e65307604abb5ad556 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921)
<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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790614226
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -36,10 +36,15 @@
/** A package private class to wrap {@link Deserializer}. */
class KafkaValueOnlyDeserializerWrapper<T> implements KafkaRecordDeserializationSchema<T> {
+
private static final long serialVersionUID = 5409547407386004054L;
+
private static final Logger LOG =
LoggerFactory.getLogger(KafkaValueOnlyDeserializerWrapper.class);
+
private final Class<? extends Deserializer<T>> deserializerClass;
+ // always be false since this Deserializer is only used for value.
Review comment:
It must always be false, since this class is only used for Kafka value. Otherwise the name this class should be changed.
--
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
[GitHub] [flink] fapaul commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
fapaul commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r788961891
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -62,8 +75,15 @@ public void open(DeserializationSchema.InitializationContext context) throws Exc
deserializerClass.getName(),
Deserializer.class,
getClass().getClassLoader());
+
+ if (config.isEmpty()) {
+ return;
+ }
+
if (deserializer instanceof Configurable) {
((Configurable) deserializer).configure(config);
+ } else {
Review comment:
Can you also add a test to verify the configuration of a de/serializer not implementing `Configurable`?
--
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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r788965796
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -62,8 +75,15 @@ public void open(DeserializationSchema.InitializationContext context) throws Exc
deserializerClass.getName(),
Deserializer.class,
getClass().getClassLoader());
+
+ if (config.isEmpty()) {
+ return;
+ }
+
if (deserializer instanceof Configurable) {
((Configurable) deserializer).configure(config);
+ } else {
Review comment:
Do you mean if it did implement Configurable, then `Deserializer.configure` will be called?
--
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
[GitHub] [flink] lindong28 commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r789248928
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSerializerWrapper.java
##########
@@ -65,8 +77,15 @@ public void open(InitializationContext context) throws Exception {
serializerClass.getName(),
Serializer.class,
getClass().getClassLoader());
+
+ if (config.isEmpty()) {
Review comment:
Is it possible that user's deserializer::configure(...) could still be useful (e.g. has side effects) even when the config is empty?
Would it be simpler and safer to remove this check?
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSerializerWrapper.java
##########
@@ -33,25 +33,37 @@
import static org.apache.flink.util.Preconditions.checkState;
class KafkaSerializerWrapper<IN> implements SerializationSchema<IN> {
+
private final Class<? extends Serializer<? super IN>> serializerClass;
+ // whether the serializer is for key or value
+ private final boolean isKey;
private final Map<String, String> config;
private final Function<? super IN, String> topicSelector;
private transient Serializer<? super IN> serializer;
KafkaSerializerWrapper(
Class<? extends Serializer<? super IN>> serializerClass,
+ boolean isKey,
Map<String, String> config,
Function<? super IN, String> topicSelector) {
this.serializerClass = checkNotNull(serializerClass);
+ this.isKey = isKey;
this.config = checkNotNull(config);
this.topicSelector = checkNotNull(topicSelector);
}
+ KafkaSerializerWrapper(
+ Class<? extends Serializer<? super IN>> serializerClass,
+ Map<String, String> config,
+ Function<? super IN, String> topicSelector) {
+ this(serializerClass, false, config, topicSelector);
+ }
+
KafkaSerializerWrapper(
Class<? extends Serializer<? super IN>> serializerClass,
Function<? super IN, String> topicSelector) {
- this(serializerClass, Collections.emptyMap(), topicSelector);
+ this(serializerClass, false, Collections.emptyMap(), topicSelector);
Review comment:
This constructor is invoked by `KafkaRecordSerializationSchemaBuilder::setKafkaKeySerializer(...)`. It is probably not intuitive to set `isKey=false` here.
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaRecordDeserializationSchema.java
##########
@@ -123,9 +121,8 @@ default void open(DeserializationSchema.InitializationContext context) throws Ex
* @return A {@link KafkaRecordDeserializationSchema} that deserialize the value with the given
* deserializer.
*/
- static <V, D extends Configurable & Deserializer<V>>
- KafkaRecordDeserializationSchema<V> valueOnly(
- Class<D> valueDeserializerClass, Map<String, String> config) {
+ static <V, D extends Deserializer<V>> KafkaRecordDeserializationSchema<V> valueOnly(
Review comment:
@fapaul I think the best way is to use `Deserializer::configure(...)` to configure deserializer. And I believe all the deserializers provided off-the-shelf by Apache Kafka, and most user-implemented deserializer, use `Deserializer::configure(...)` instead of `Configurable`.
`Configurable::configure(...)` is not deprecated in Kafka and I believe Flink still supports this for backward compatibility. And I agree it is useful to extend the functionality of this method to support `Deserializer::configure(...)` as long as we could make sure we don't break users existing application.
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaRecordDeserializationSchema.java
##########
@@ -123,9 +121,8 @@ default void open(DeserializationSchema.InitializationContext context) throws Ex
* @return A {@link KafkaRecordDeserializationSchema} that deserialize the value with the given
* deserializer.
*/
- static <V, D extends Configurable & Deserializer<V>>
- KafkaRecordDeserializationSchema<V> valueOnly(
- Class<D> valueDeserializerClass, Map<String, String> config) {
+ static <V, D extends Deserializer<V>> KafkaRecordDeserializationSchema<V> valueOnly(
Review comment:
@JingGe I see. I agree it is still backward compatible (suppose we remove the `config.isEmpty()` check) if user invokes this method with a `Configurable` instance.
It seems that the Java doc of this method still says `only valid when the deserializer is an implementation of {@code Configurable}`. Do we need to update 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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dc25efa3f7f3488c227b79c99d5639037ed35655 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664)
* 89187e8fe1fd38cf50e7f2188cc5cd14c0b65089 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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790754460
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -36,10 +36,15 @@
/** A package private class to wrap {@link Deserializer}. */
class KafkaValueOnlyDeserializerWrapper<T> implements KafkaRecordDeserializationSchema<T> {
+
private static final long serialVersionUID = 5409547407386004054L;
+
private static final Logger LOG =
LoggerFactory.getLogger(KafkaValueOnlyDeserializerWrapper.class);
+
private final Class<? extends Deserializer<T>> deserializerClass;
+ // always be false since this Deserializer is only used for value.
Review comment:
It is because of code style consistency and don't hide the magic deeply into the code at the corner.
--
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
[GitHub] [flink] lindong28 closed pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
lindong28 closed pull request #18397:
URL: https://github.com/apache/flink/pull/18397
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 89187e8fe1fd38cf50e7f2188cc5cd14c0b65089 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714)
* 2c46519978e62a0f60b8a178d126458d3c11da4b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912)
* 386e9d632753391d80bbbcd41dacb6a1786d5727 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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
}, {
"hash" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921",
"triggerID" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"triggerType" : "PUSH"
}, {
"hash" : "3e1198a17fea03035bc621268a39ea3b386809a4",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30030",
"triggerID" : "3e1198a17fea03035bc621268a39ea3b386809a4",
"triggerType" : "PUSH"
}, {
"hash" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5b2d44dfea90aa64e7f9c2e65307604abb5ad556 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921)
* 3e1198a17fea03035bc621268a39ea3b386809a4 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30030)
* 1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87 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
[GitHub] [flink] lindong28 closed pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
lindong28 closed pull request #18397:
URL: https://github.com/apache/flink/pull/18397
--
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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r789714596
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -62,8 +75,15 @@ public void open(DeserializationSchema.InitializationContext context) throws Exc
deserializerClass.getName(),
Deserializer.class,
getClass().getClassLoader());
+
+ if (config.isEmpty()) {
+ return;
+ }
+
if (deserializer instanceof Configurable) {
((Configurable) deserializer).configure(config);
+ } else {
Review comment:
The first one was there before this PR, it should already be tested. For the third one, either we do nothing like I proposed which might need a test, or we don't change it, which will be one of the first or the second case.
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 89187e8fe1fd38cf50e7f2188cc5cd14c0b65089 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714)
* 2c46519978e62a0f60b8a178d126458d3c11da4b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912)
* 386e9d632753391d80bbbcd41dacb6a1786d5727 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916)
<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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [HOTFIX][Kafka] update interface @PublicEvolving or @Internal
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dc25efa3f7f3488c227b79c99d5639037ed35655 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664)
<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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790754460
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -36,10 +36,15 @@
/** A package private class to wrap {@link Deserializer}. */
class KafkaValueOnlyDeserializerWrapper<T> implements KafkaRecordDeserializationSchema<T> {
+
private static final long serialVersionUID = 5409547407386004054L;
+
private static final Logger LOG =
LoggerFactory.getLogger(KafkaValueOnlyDeserializerWrapper.class);
+
private final Class<? extends Deserializer<T>> deserializerClass;
+ // always be false since this Deserializer is only used for value.
Review comment:
It is because of code style consistency and don't hiding the magic deeply into the code at the corner. I did it like your suggestion previously and changed to the current solution.
--
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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790614226
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -36,10 +36,15 @@
/** A package private class to wrap {@link Deserializer}. */
class KafkaValueOnlyDeserializerWrapper<T> implements KafkaRecordDeserializationSchema<T> {
+
private static final long serialVersionUID = 5409547407386004054L;
+
private static final Logger LOG =
LoggerFactory.getLogger(KafkaValueOnlyDeserializerWrapper.class);
+
private final Class<? extends Deserializer<T>> deserializerClass;
+ // always be false since this Deserializer is only used for value.
Review comment:
It must always be false, since this class is only used for Kafka value. Otherwise the name of this class should be changed.
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
}, {
"hash" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921",
"triggerID" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"triggerType" : "PUSH"
}, {
"hash" : "3e1198a17fea03035bc621268a39ea3b386809a4",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30030",
"triggerID" : "3e1198a17fea03035bc621268a39ea3b386809a4",
"triggerType" : "PUSH"
}, {
"hash" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30080",
"triggerID" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5b2d44dfea90aa64e7f9c2e65307604abb5ad556 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921)
* 3e1198a17fea03035bc621268a39ea3b386809a4 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30030)
* 1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30080)
<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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
}, {
"hash" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921",
"triggerID" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"triggerType" : "PUSH"
}, {
"hash" : "3e1198a17fea03035bc621268a39ea3b386809a4",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30030",
"triggerID" : "3e1198a17fea03035bc621268a39ea3b386809a4",
"triggerType" : "PUSH"
}, {
"hash" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30080",
"triggerID" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"triggerType" : "PUSH"
}, {
"hash" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30080",
"triggerID" : "1020716290",
"triggerType" : "MANUAL"
}, {
"hash" : "1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30080",
"triggerID" : "1020958323",
"triggerType" : "MANUAL"
} ]
}-->
## CI report:
* 1cc4394ed63adc3729cc13c2ef7aebebdf0d4e87 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=30080)
<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
[GitHub] [flink] JingGe commented on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1020958184
Many thanks for the review!
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
}, {
"hash" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921",
"triggerID" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5b2d44dfea90aa64e7f9c2e65307604abb5ad556 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29921)
<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
[GitHub] [flink] lindong28 commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r788860986
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaRecordDeserializationSchema.java
##########
@@ -123,9 +121,8 @@ default void open(DeserializationSchema.InitializationContext context) throws Ex
* @return A {@link KafkaRecordDeserializationSchema} that deserialize the value with the given
* deserializer.
*/
- static <V, D extends Configurable & Deserializer<V>>
- KafkaRecordDeserializationSchema<V> valueOnly(
- Class<D> valueDeserializerClass, Map<String, String> config) {
+ static <V, D extends Deserializer<V>> KafkaRecordDeserializationSchema<V> valueOnly(
Review comment:
Prior to https://github.com/apache/flink/pull/15051, we have the following method in the class `KafkaRecordDeserializer`:
```
/**
* Wraps a Kafka {@link Deserializer} to a {@link KafkaRecordDeserializer}.
*
* @param valueDeserializerClass the deserializer class used to deserialize the value.
* @param config the configuration of the value deserializer, only valid when the deserializer
* is an implementation of {@code Configurable}.
* @param <V> the value type.
* @param <D> the type of the deserializer.
* @return A {@link KafkaRecordDeserializer} that deserialize the value with the given
* deserializer.
*/
static <V, D extends Configurable & Deserializer<V>> KafkaRecordDeserializer<V> valueOnly(
Class<D> valueDeserializerClass, Map<String, String> config) {
...
}
```
This method basically assumes that users might have defined a deserializerA as a subclass of both Configurable and Deserializer, where `deserializerA::configure(Map<String, ?>)` could have user-defined configuration logic. Since this subclass could be defined by user outside the Flink project, we want to continue to support this this user-defined deserializer for backward compatibility.
Changing the method signature in this way could cause backward incompatibility. Though this method is not marked `@public`, I think this is because we forgot to add this tag. Given that this method is supposed to be used by public users, I feel it is preferred not to break it.
What do you think?
--
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
[GitHub] [flink] lindong28 commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r788865671
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaRecordDeserializationSchema.java
##########
@@ -123,9 +121,8 @@ default void open(DeserializationSchema.InitializationContext context) throws Ex
* @return A {@link KafkaRecordDeserializationSchema} that deserialize the value with the given
* deserializer.
*/
- static <V, D extends Configurable & Deserializer<V>>
- KafkaRecordDeserializationSchema<V> valueOnly(
- Class<D> valueDeserializerClass, Map<String, String> config) {
+ static <V, D extends Deserializer<V>> KafkaRecordDeserializationSchema<V> valueOnly(
Review comment:
BTW, this method (prior to https://github.com/apache/flink/pull/15051) was added in FLINK-18323. Jiangjie Qin, author of that PR, was pretty familiar with Kafka. I believe he has good reason to add that method, which uses `org.apache.kafka.common.Configurable.configure` instead of `org.apache.kafka.common.serialization.Deserializer.configure`. And I guess the reason is likely due to backward compatibility.
--
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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790964213
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilderTest.java
##########
@@ -160,6 +165,37 @@ public void testSerializeRecordWithKey() {
assertArrayEquals(record.value(), serializationSchema.serialize("a"));
}
+ @Test
+ public void testNoConfigurableKafkaKeySerializer() throws Exception {
+ final Map<String, String> config = ImmutableMap.of("simpleKey", "simpleValue");
+ final KafkaRecordSerializationSchema<String> schema =
+ KafkaRecordSerializationSchema.builder()
+ .setTopic(DEFAULT_TOPIC)
+ // use StringSerializer as dummy Serializer, since ValueSerializer is
Review comment:
I'll change it. Just out of curiosity, I didn't find the consistency rule of single line comment with the first letter being capital. There are also many opposite examples using lowercase in the codebase, e.g. https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/StreamFormatAdapter.java
https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
--
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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790964213
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilderTest.java
##########
@@ -160,6 +165,37 @@ public void testSerializeRecordWithKey() {
assertArrayEquals(record.value(), serializationSchema.serialize("a"));
}
+ @Test
+ public void testNoConfigurableKafkaKeySerializer() throws Exception {
+ final Map<String, String> config = ImmutableMap.of("simpleKey", "simpleValue");
+ final KafkaRecordSerializationSchema<String> schema =
+ KafkaRecordSerializationSchema.builder()
+ .setTopic(DEFAULT_TOPIC)
+ // use StringSerializer as dummy Serializer, since ValueSerializer is
Review comment:
I'll change it. Just out of curiosity, I didn't find the consistency rule of single line comment with the first letter being capital. There are also many opposite examples in the codebase, e.g. https://github.com/JingGe/flink/blob/master/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/StreamFormatAdapter.java
https://github.com/JingGe/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 89187e8fe1fd38cf50e7f2188cc5cd14c0b65089 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714)
* 2c46519978e62a0f60b8a178d126458d3c11da4b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912)
* 386e9d632753391d80bbbcd41dacb6a1786d5727 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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 89187e8fe1fd38cf50e7f2188cc5cd14c0b65089 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714)
* 2c46519978e62a0f60b8a178d126458d3c11da4b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912)
<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
[GitHub] [flink] lindong28 commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790745777
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -36,10 +36,15 @@
/** A package private class to wrap {@link Deserializer}. */
class KafkaValueOnlyDeserializerWrapper<T> implements KafkaRecordDeserializationSchema<T> {
+
private static final long serialVersionUID = 5409547407386004054L;
+
private static final Logger LOG =
LoggerFactory.getLogger(KafkaValueOnlyDeserializerWrapper.class);
+
private final Class<? extends Deserializer<T>> deserializerClass;
+ // always be false since this Deserializer is only used for value.
Review comment:
I see why its value is always false. Since its value is always false, would it be simpler to just use `deserializer.configure(config, false)`? We could add a comment above if needed.
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 89187e8fe1fd38cf50e7f2188cc5cd14c0b65089 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714)
* 2c46519978e62a0f60b8a178d126458d3c11da4b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912)
<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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
--
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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 89187e8fe1fd38cf50e7f2188cc5cd14c0b65089 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714)
* 2c46519978e62a0f60b8a178d126458d3c11da4b 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
[GitHub] [flink] flinkbot edited a comment on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29664",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
}, {
"hash" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714",
"triggerID" : "89187e8fe1fd38cf50e7f2188cc5cd14c0b65089",
"triggerType" : "PUSH"
}, {
"hash" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912",
"triggerID" : "2c46519978e62a0f60b8a178d126458d3c11da4b",
"triggerType" : "PUSH"
}, {
"hash" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916",
"triggerID" : "386e9d632753391d80bbbcd41dacb6a1786d5727",
"triggerType" : "PUSH"
}, {
"hash" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5b2d44dfea90aa64e7f9c2e65307604abb5ad556",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 89187e8fe1fd38cf50e7f2188cc5cd14c0b65089 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29714)
* 2c46519978e62a0f60b8a178d126458d3c11da4b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29912)
* 386e9d632753391d80bbbcd41dacb6a1786d5727 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=29916)
* 5b2d44dfea90aa64e7f9c2e65307604abb5ad556 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
[GitHub] [flink] JingGe commented on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1017220934
@lindong28 @fapaul Would you like to take a look at this PR? Many thanks for your effort!
--
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
[GitHub] [flink] lindong28 commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r788868076
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaRecordDeserializationSchema.java
##########
@@ -123,9 +121,8 @@ default void open(DeserializationSchema.InitializationContext context) throws Ex
* @return A {@link KafkaRecordDeserializationSchema} that deserialize the value with the given
* deserializer.
*/
- static <V, D extends Configurable & Deserializer<V>>
- KafkaRecordDeserializationSchema<V> valueOnly(
- Class<D> valueDeserializerClass, Map<String, String> config) {
+ static <V, D extends Deserializer<V>> KafkaRecordDeserializationSchema<V> valueOnly(
Review comment:
BTW, this method (prior to https://github.com/apache/flink/pull/15051) was added in FLINK-18323. Jiangjie Qin, author of that PR, was pretty familiar with Kafka. I believe he has good reason to add that method, which uses `org.apache.kafka.common.Configurable.configure` instead of `org.apache.kafka.common.serialization.Deserializer.configure`. And I guess the reason is likely due to backward compatibility.
--
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
[GitHub] [flink] flinkbot commented on pull request #18397: [HOTFIX][Kafka] update interface @PublicEvolving or @Internal
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1015848205
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "dc25efa3f7f3488c227b79c99d5639037ed35655",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dc25efa3f7f3488c227b79c99d5639037ed35655 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
[GitHub] [flink] lindong28 commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790745777
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -36,10 +36,15 @@
/** A package private class to wrap {@link Deserializer}. */
class KafkaValueOnlyDeserializerWrapper<T> implements KafkaRecordDeserializationSchema<T> {
+
private static final long serialVersionUID = 5409547407386004054L;
+
private static final Logger LOG =
LoggerFactory.getLogger(KafkaValueOnlyDeserializerWrapper.class);
+
private final Class<? extends Deserializer<T>> deserializerClass;
+ // always be false since this Deserializer is only used for value.
Review comment:
I see why its value is always false. Since its value is always false, would it be simpler to just use `deserializer.configure(config, false)` instead of defining this variable? We could add a comment above if needed.
--
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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790964213
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilderTest.java
##########
@@ -160,6 +165,37 @@ public void testSerializeRecordWithKey() {
assertArrayEquals(record.value(), serializationSchema.serialize("a"));
}
+ @Test
+ public void testNoConfigurableKafkaKeySerializer() throws Exception {
+ final Map<String, String> config = ImmutableMap.of("simpleKey", "simpleValue");
+ final KafkaRecordSerializationSchema<String> schema =
+ KafkaRecordSerializationSchema.builder()
+ .setTopic(DEFAULT_TOPIC)
+ // use StringSerializer as dummy Serializer, since ValueSerializer is
Review comment:
I'll change it. Just out of curiosity, I didn't find the consistency rule of single line comment with the first letter being capital. There are also many opposite example in the codebase, e.g. https://github.com/JingGe/flink/blob/master/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/StreamFormatAdapter.java
--
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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790956272
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaRecordDeserializationSchemaTest.java
##########
@@ -93,6 +112,29 @@ public void testKafkaValueDeserializerWrapper() throws Exception {
assertEquals("world", collector.list.get(0));
}
+ @Test
+ public void testNoConfigurableKafkaSerializer() throws Exception {
Review comment:
ops, that was a mistake done by c&p, 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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790939938
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -36,10 +36,15 @@
/** A package private class to wrap {@link Deserializer}. */
class KafkaValueOnlyDeserializerWrapper<T> implements KafkaRecordDeserializationSchema<T> {
+
private static final long serialVersionUID = 5409547407386004054L;
+
private static final Logger LOG =
LoggerFactory.getLogger(KafkaValueOnlyDeserializerWrapper.class);
+
private final Class<? extends Deserializer<T>> deserializerClass;
+ // always be false since this Deserializer is only used for value.
Review comment:
Code style consistency means in this case that developers, when they come from `KafkaSerializerWrapper` will check the `isKey` at first to see if `deserializer.configure(...)` has been used in `KafkaValueOnlyDeserializerWrapper`. It is much easier to check it via instance variable than digging into the code. Try to read both classes back and forth, you will get the feeling.
It is not a big deal. I will change it back.
--
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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790964213
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilderTest.java
##########
@@ -160,6 +165,37 @@ public void testSerializeRecordWithKey() {
assertArrayEquals(record.value(), serializationSchema.serialize("a"));
}
+ @Test
+ public void testNoConfigurableKafkaKeySerializer() throws Exception {
+ final Map<String, String> config = ImmutableMap.of("simpleKey", "simpleValue");
+ final KafkaRecordSerializationSchema<String> schema =
+ KafkaRecordSerializationSchema.builder()
+ .setTopic(DEFAULT_TOPIC)
+ // use StringSerializer as dummy Serializer, since ValueSerializer is
Review comment:
I'll change it. Just out of curiosity, I didn't find the consistency rule of single line comment with the first letter being capital. There are also many opposite examples using lowercase in the codebase, e.g. https://github.com/JingGe/flink/blob/master/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/StreamFormatAdapter.java
https://github.com/JingGe/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
--
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
[GitHub] [flink] JingGe commented on pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on pull request #18397:
URL: https://github.com/apache/flink/pull/18397#issuecomment-1020958184
--
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
[GitHub] [flink] JingGe commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
JingGe commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r790939938
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -36,10 +36,15 @@
/** A package private class to wrap {@link Deserializer}. */
class KafkaValueOnlyDeserializerWrapper<T> implements KafkaRecordDeserializationSchema<T> {
+
private static final long serialVersionUID = 5409547407386004054L;
+
private static final Logger LOG =
LoggerFactory.getLogger(KafkaValueOnlyDeserializerWrapper.class);
+
private final Class<? extends Deserializer<T>> deserializerClass;
+ // always be false since this Deserializer is only used for value.
Review comment:
Code style consistency means in this case that developers, when the come from `KafkaSerializerWrapper` will check the `isKey` at first to see if `deserializer.configure(...)` has been used in `KafkaValueOnlyDeserializerWrapper`. It is much easier to check it via instance variable than digging into the code. Try to read both classes back and forth, you will get the feeling.
It is not a big deal. I will change it back.
--
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
[GitHub] [flink] lindong28 commented on a change in pull request #18397: [FLINK-25702][Kafka] Use the configure feature provided by the kafka Serializer/Deserializer.
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #18397:
URL: https://github.com/apache/flink/pull/18397#discussion_r791272806
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java
##########
@@ -36,10 +36,15 @@
/** A package private class to wrap {@link Deserializer}. */
class KafkaValueOnlyDeserializerWrapper<T> implements KafkaRecordDeserializationSchema<T> {
+
private static final long serialVersionUID = 5409547407386004054L;
+
private static final Logger LOG =
LoggerFactory.getLogger(KafkaValueOnlyDeserializerWrapper.class);
+
private final Class<? extends Deserializer<T>> deserializerClass;
+ // always be false since this Deserializer is only used for value.
Review comment:
Thank you Jing for digging into 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