You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "davidradl (via GitHub)" <gi...@apache.org> on 2025/12/18 09:50:22 UTC
[PR] [FLINK-38816] use Flink 2 compatible deserialise [flink-connector-http]
davidradl opened a new pull request, #12:
URL: https://github.com/apache/flink-connector-http/pull/12
The version of deserialise method that is used by this connector is deprecated at Flink 2, and a Runtime Exception is thrown. This change means that the Flink 2 compatible deserialise method is now used.
This is a port of the GetInData PR [HTTP 204](https://github.com/getindata/flink-http-connector/pull/204); getindata requires 90% code coverage, the code has been implemented to be testable and meet this requirement then ported as is.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38816] use Flink 2 compatible deserialise [flink-connector-http]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on PR #12:
URL: https://github.com/apache/flink-connector-http/pull/12#issuecomment-3729310173
changing to draft while I check something out
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38816] use Flink 2 compatible deserialise [flink-connector-http]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #12:
URL: https://github.com/apache/flink-connector-http/pull/12#discussion_r2675928554
##########
flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java:
##########
@@ -326,22 +327,49 @@ private Collection<RowData> deserialize(String responseBody) throws IOException
}
}
- private List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
- return Optional.ofNullable(responseBodyDecoder.deserialize(rawBytes))
- .map(Collections::singletonList)
- .orElse(Collections.emptyList());
+ @VisibleForTesting
+ List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
+ List<RowData> result = new ArrayList<>();
+ responseBodyDecoder.deserialize(rawBytes, createRowDataCollector(result));
+ return result;
+ }
+
+ @VisibleForTesting
+ Collector<RowData> createRowDataCollector(List<RowData> result) {
+ return new RowDataCollector(result);
+ }
+
+ /** A simple collector implementation that adds RowData records to a list. */
+ @VisibleForTesting
+ static class RowDataCollector implements Collector<RowData> {
+ private final List<RowData> result;
+
+ RowDataCollector(List<RowData> result) {
+ this.result = result;
+ }
+
+ @Override
+ public void collect(RowData record) {
+ result.add(record);
+ }
+
+ @Override
+ public void close() {
+ // No-op - nothing to clean up
+ }
}
- private List<RowData> deserializeArray(byte[] rawBytes) throws IOException {
+ @VisibleForTesting
Review Comment:
I think we should keep this @VisibleForTesting as it justifies why this method is not private.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38816] use Flink 2 compatible deserialise [flink-connector-http]
Posted by "ferenc-csaky (via GitHub)" <gi...@apache.org>.
ferenc-csaky commented on code in PR #12:
URL: https://github.com/apache/flink-connector-http/pull/12#discussion_r2677051937
##########
flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java:
##########
@@ -327,21 +328,22 @@ private Collection<RowData> deserialize(String responseBody) throws IOException
}
private List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
- return Optional.ofNullable(responseBodyDecoder.deserialize(rawBytes))
- .map(Collections::singletonList)
- .orElse(Collections.emptyList());
+ List<RowData> result = new ArrayList<>();
+ responseBodyDecoder.deserialize(rawBytes, new ListCollector(result));
+ return result;
Review Comment:
One thing I missed before, let's wrap it to `Collections.unmodifiableList(result)`, cause the prev. code returned an immutable list.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38816] use Flink 2 compatible deserialise [flink-connector-http]
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #12:
URL: https://github.com/apache/flink-connector-http/pull/12#discussion_r2673059410
##########
flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java:
##########
@@ -326,22 +327,49 @@ private Collection<RowData> deserialize(String responseBody) throws IOException
}
}
- private List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
- return Optional.ofNullable(responseBodyDecoder.deserialize(rawBytes))
- .map(Collections::singletonList)
- .orElse(Collections.emptyList());
+ @VisibleForTesting
+ List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
+ List<RowData> result = new ArrayList<>();
+ responseBodyDecoder.deserialize(rawBytes, createRowDataCollector(result));
+ return result;
+ }
+
+ @VisibleForTesting
+ Collector<RowData> createRowDataCollector(List<RowData> result) {
+ return new RowDataCollector(result);
+ }
+
+ /** A simple collector implementation that adds RowData records to a list. */
+ @VisibleForTesting
Review Comment:
ditto.
##########
flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java:
##########
@@ -326,22 +327,49 @@ private Collection<RowData> deserialize(String responseBody) throws IOException
}
}
- private List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
- return Optional.ofNullable(responseBodyDecoder.deserialize(rawBytes))
- .map(Collections::singletonList)
- .orElse(Collections.emptyList());
+ @VisibleForTesting
Review Comment:
could the line be deleted?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38816] use Flink 2 compatible deserialise [flink-connector-http]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on PR #12:
URL: https://github.com/apache/flink-connector-http/pull/12#issuecomment-3718114230
> I am not sure I understand this change. The `DeserializationSchema` interface is exactly the same in [Flink 1.20](https://github.com/apache/flink/blob/release-1.20/flink-core/src/main/java/org/apache/flink/api/common/serialization/DeserializationSchema.java) and the current [master](https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/api/common/serialization/DeserializationSchema.java), so I do not see what this solves.
@ferenc-csaky It is the [https://github.com/apache/flink/blob/fec9336e7d99500aeb9097e441d8da0e6bde5943/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/AbstractJsonDeserializationSchema.java#L115](https://github.com/apache/flink/blob/fec9336e7d99500aeb9097e441d8da0e6bde5943/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/AbstractJsonDeserializationSchema.java#L115) that is deprecated - I should have mentioned this class by name for clarity. You can see that at version 2 the deprecated method does not continue to work, as you might expect, it throws an Exception.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38816] use Flink 2 compatible deserialise [flink-connector-http]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on PR #12:
URL: https://github.com/apache/flink-connector-http/pull/12#issuecomment-3669408591
@ferenc-csaky Please could you have a look at this one?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38816] use Flink 2 compatible deserialise [flink-connector-http]
Posted by "ferenc-csaky (via GitHub)" <gi...@apache.org>.
ferenc-csaky merged PR #12:
URL: https://github.com/apache/flink-connector-http/pull/12
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38816] use Flink 2 compatible deserialise [flink-connector-http]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #12:
URL: https://github.com/apache/flink-connector-http/pull/12#discussion_r2675920860
##########
flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java:
##########
@@ -326,22 +327,49 @@ private Collection<RowData> deserialize(String responseBody) throws IOException
}
}
- private List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
- return Optional.ofNullable(responseBodyDecoder.deserialize(rawBytes))
- .map(Collections::singletonList)
- .orElse(Collections.emptyList());
+ @VisibleForTesting
Review Comment:
With the otehr change I agree - I will remove this tag and make the method private
##########
flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java:
##########
@@ -326,22 +327,49 @@ private Collection<RowData> deserialize(String responseBody) throws IOException
}
}
- private List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
- return Optional.ofNullable(responseBodyDecoder.deserialize(rawBytes))
- .map(Collections::singletonList)
- .orElse(Collections.emptyList());
+ @VisibleForTesting
Review Comment:
With the other change I agree - I will remove this tag and make the method private
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38816] use Flink 2 compatible deserialise [flink-connector-http]
Posted by "ferenc-csaky (via GitHub)" <gi...@apache.org>.
ferenc-csaky commented on PR #12:
URL: https://github.com/apache/flink-connector-http/pull/12#issuecomment-3715555497
I am not sure I understand this change. The `DeserializationSchema` interface is exactly the same in [Flink 1.20](https://github.com/apache/flink/blob/release-1.20/flink-core/src/main/java/org/apache/flink/api/common/serialization/DeserializationSchema.java) and the current [master](https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/api/common/serialization/DeserializationSchema.java), so I do not see what this solves.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38816] use Flink 2 compatible deserialise [flink-connector-http]
Posted by "ferenc-csaky (via GitHub)" <gi...@apache.org>.
ferenc-csaky commented on code in PR #12:
URL: https://github.com/apache/flink-connector-http/pull/12#discussion_r2665617620
##########
flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java:
##########
@@ -326,22 +327,49 @@ private Collection<RowData> deserialize(String responseBody) throws IOException
}
}
- private List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
- return Optional.ofNullable(responseBodyDecoder.deserialize(rawBytes))
- .map(Collections::singletonList)
- .orElse(Collections.emptyList());
+ @VisibleForTesting
+ List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
+ List<RowData> result = new ArrayList<>();
+ responseBodyDecoder.deserialize(rawBytes, createRowDataCollector(result));
+ return result;
+ }
+
+ @VisibleForTesting
+ Collector<RowData> createRowDataCollector(List<RowData> result) {
+ return new RowDataCollector(result);
+ }
+
+ /** A simple collector implementation that adds RowData records to a list. */
+ @VisibleForTesting
+ static class RowDataCollector implements Collector<RowData> {
Review Comment:
There is a `ListCollector` which does exactly this, so we can get rid of this class.
##########
flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java:
##########
@@ -326,22 +327,49 @@ private Collection<RowData> deserialize(String responseBody) throws IOException
}
}
- private List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
- return Optional.ofNullable(responseBodyDecoder.deserialize(rawBytes))
- .map(Collections::singletonList)
- .orElse(Collections.emptyList());
+ @VisibleForTesting
+ List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
+ List<RowData> result = new ArrayList<>();
+ responseBodyDecoder.deserialize(rawBytes, createRowDataCollector(result));
+ return result;
+ }
+
+ @VisibleForTesting
+ Collector<RowData> createRowDataCollector(List<RowData> result) {
+ return new RowDataCollector(result);
+ }
Review Comment:
Any specific reason for `createRowDataCollector`? I do not see why not call the ctor directly in `deserializeSingleValue`:
```java
List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
List<RowData> result = new ArrayList<>();
responseBodyDecoder.deserialize(rawBytes, new ListCollector(result));
return result;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38816] use Flink 2 compatible deserialise [flink-connector-http]
Posted by "ferenc-csaky (via GitHub)" <gi...@apache.org>.
ferenc-csaky commented on code in PR #12:
URL: https://github.com/apache/flink-connector-http/pull/12#discussion_r2682129861
##########
flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java:
##########
@@ -326,22 +326,42 @@ private Collection<RowData> deserialize(String responseBody) throws IOException
}
}
- private List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
- return Optional.ofNullable(responseBodyDecoder.deserialize(rawBytes))
- .map(Collections::singletonList)
- .orElse(Collections.emptyList());
+ @VisibleForTesting
+ List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
+ List<RowData> result = new ArrayList<>();
+ responseBodyDecoder.deserialize(rawBytes, new ListCollector(result));
+ return Collections.unmodifiableList(result);
}
- private List<RowData> deserializeArray(byte[] rawBytes) throws IOException {
+ private static class ListCollector implements org.apache.flink.util.Collector<RowData> {
Review Comment:
I just merged it and spotted this came back. Why?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38816] use Flink 2 compatible deserialise [flink-connector-http]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #12:
URL: https://github.com/apache/flink-connector-http/pull/12#discussion_r2682230649
##########
flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java:
##########
@@ -326,22 +326,42 @@ private Collection<RowData> deserialize(String responseBody) throws IOException
}
}
- private List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
- return Optional.ofNullable(responseBodyDecoder.deserialize(rawBytes))
- .map(Collections::singletonList)
- .orElse(Collections.emptyList());
+ @VisibleForTesting
+ List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
+ List<RowData> result = new ArrayList<>();
+ responseBodyDecoder.deserialize(rawBytes, new ListCollector(result));
+ return Collections.unmodifiableList(result);
}
- private List<RowData> deserializeArray(byte[] rawBytes) throws IOException {
+ private static class ListCollector implements org.apache.flink.util.Collector<RowData> {
Review Comment:
@ferenc-csaky My mistake - I meant this to be left unchanged - let me raise another PR to address this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38816] use Flink 2 compatible deserialise [flink-connector-http]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #12:
URL: https://github.com/apache/flink-connector-http/pull/12#discussion_r2681955746
##########
.gitignore:
##########
@@ -5,6 +5,7 @@
.gitignore.swp
.project
.settings
+.metals
Review Comment:
my IDE added this so put in an ignore for it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38816] use Flink 2 compatible deserialise [flink-connector-http]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #12:
URL: https://github.com/apache/flink-connector-http/pull/12#discussion_r2675925639
##########
flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/JavaNetHttpPollingClient.java:
##########
@@ -326,22 +327,49 @@ private Collection<RowData> deserialize(String responseBody) throws IOException
}
}
- private List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
- return Optional.ofNullable(responseBodyDecoder.deserialize(rawBytes))
- .map(Collections::singletonList)
- .orElse(Collections.emptyList());
+ @VisibleForTesting
+ List<RowData> deserializeSingleValue(byte[] rawBytes) throws IOException {
+ List<RowData> result = new ArrayList<>();
+ responseBodyDecoder.deserialize(rawBytes, createRowDataCollector(result));
+ return result;
+ }
+
+ @VisibleForTesting
+ Collector<RowData> createRowDataCollector(List<RowData> result) {
+ return new RowDataCollector(result);
+ }
+
+ /** A simple collector implementation that adds RowData records to a list. */
+ @VisibleForTesting
Review Comment:
this class will be removed with the simplication Ferenc suggested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38816] use Flink 2 compatible deserialise [flink-connector-http]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on PR #12:
URL: https://github.com/apache/flink-connector-http/pull/12#issuecomment-3729258596
@RocMarshal @ferenc-csaky thanks for your feedback - it is ready for review again.
--
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