You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "hugogu (via GitHub)" <gi...@apache.org> on 2024/04/03 10:14:18 UTC
[PR] [FLINK-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
hugogu opened a new pull request, #89:
URL: https://github.com/apache/flink-connector-kafka/pull/89
So that when connector-kafka installed into Flink Libs, it can work with a Customer Deserializer class defines in User Code.
This is useful with work with Spring Kafka JsonDeserializer. JsonDeserializer is a generic class and can't be used to build a KafkaSource directly because the Generic Type information required would lost.
A typical use case of JsonDeserializer looks like following
* Declare a sub class of JsonDeserializer with customized object mapper (which is in User Jar for sure)
```kotlin
class AccountBalanceActivityDeserializer :
JsonDeserializer<AccountBalanceActivity>(SpringObjectMapperSupplier.INSTANCE)
```
* Use it to build a KafkaSource
```kotlin
@Bean
fun source(): KafkaSource<AccountBalanceActivity> {
return KafkaSource.builder<AccountBalanceActivity>()
.setBootstrapServers("kafka:9092")
.setTopics("BalanceActivity")
.setGroupId("FraudDetection")
.setStartingOffsets(OffsetsInitializer.latest())
.setDeserializer(KafkaRecordDeserializationSchema.valueOnly(AccountBalanceActivityDeserializer::class.java))
.build()
}
```
This fix is to make above sample use pattern work when connector-kafka installed into Flink Libs rather than included in user jar.
--
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-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #89:
URL: https://github.com/apache/flink-connector-kafka/pull/89#issuecomment-3668600220
This PR has been closed since it has not had any activity in 120 days.
If you feel like this was a mistake, or you would like to continue working on it,
please feel free to re-open the PR and ask for a 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
Re: [PR] [FLINK-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #89:
URL: https://github.com/apache/flink-connector-kafka/pull/89#issuecomment-3545499246
This PR is being marked as stale since it has not had any activity in the last 90 days.
If you would like to keep this PR alive, please leave a comment asking for a review.
If the PR has merge conflicts, update it with the latest from the base branch.
If you are having difficulty finding a reviewer, please reach out to the
community, contact details can be found here: https://flink.apache.org/what-is-flink/community/
If this PR is no longer valid or desired, please feel free to close it.
If no activity occurs in the next 30 days, it will be automatically closed.
--
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-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #89: [FLINK-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer
URL: https://github.com/apache/flink-connector-kafka/pull/89
--
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-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on PR #89:
URL: https://github.com/apache/flink-connector-kafka/pull/89#issuecomment-2034536889
> There are existing tests covering this class and exactly this line of code and I also verified it pass. I'm happy to add more tests to ensure the user code class loader is being used.
I don't think that's correct, because if there were existing tests I would have expected those to fail because of your change.
--
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-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
Posted by "morazow (via GitHub)" <gi...@apache.org>.
morazow commented on PR #89:
URL: https://github.com/apache/flink-connector-kafka/pull/89#issuecomment-2044272245
Looks good, thanks @hugogu for quick update! Hopefully the CI is green this time 🤞
--
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-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on code in PR #89:
URL: https://github.com/apache/flink-connector-kafka/pull/89#discussion_r1549969179
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSerializerWrapperTest.java:
##########
@@ -0,0 +1,28 @@
+package org.apache.flink.connector.kafka.sink;
+
+import org.apache.flink.streaming.connectors.kafka.testutils.SerializationTestBase;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
Review Comment:
Per the code style and quality guide, please don't use Mockito https://flink.apache.org/how-to-contribute/code-style-and-quality-common/#avoid-mockito---use-reusable-test-implementations
--
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-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
Posted by "hugogu (via GitHub)" <gi...@apache.org>.
hugogu commented on PR #89:
URL: https://github.com/apache/flink-connector-kafka/pull/89#issuecomment-2034716402
@MartijnVisser May I have your advice on the use of `TemporaryClassLoaderContext` in the implementation? It looks unnecessary to me. Do you think I should just remove it in this PR?
```java
public void open(InitializationContext context) throws Exception {
final ClassLoader userCodeClassLoader = context.getUserCodeClassLoader().asClassLoader();
try (TemporaryClassLoaderContext ignored =
TemporaryClassLoaderContext.of(userCodeClassLoader)) {
serializer =
InstantiationUtil.instantiate(
serializerClass.getName(),
Serializer.class,
userCodeClassLoader);
if (serializer instanceof Configurable) {
((Configurable) serializer).configure(config);
} else {
serializer.configure(config, isKey);
}
} catch (Exception e) {
throw new IOException("Failed to instantiate the serializer of class " + serializer, e);
}
}
```
--
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-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
Posted by "hugogu (via GitHub)" <gi...@apache.org>.
hugogu commented on PR #89:
URL: https://github.com/apache/flink-connector-kafka/pull/89#issuecomment-2034169137
@MartijnVisser Thanks for the quick response. There are existing tests covering this class and exactly this line of code and I also verified it pass. I'm happy to add more tests to ensure the user code class loader is being used.
<img width="526" alt="image" src="https://github.com/apache/flink-connector-kafka/assets/2989766/f0dbbe5c-5bf4-4d22-8943-b301bca420e4">
--
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-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
Posted by "morazow (via GitHub)" <gi...@apache.org>.
morazow commented on code in PR #89:
URL: https://github.com/apache/flink-connector-kafka/pull/89#discussion_r1555308164
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSerializerWrapperTest.java:
##########
@@ -0,0 +1,63 @@
+package org.apache.flink.connector.kafka.sink;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.util.FlinkUserCodeClassLoaders;
+import org.apache.flink.util.SimpleUserCodeClassLoader;
+import org.apache.flink.util.UserCodeClassLoader;
+
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.Test;
+
+import java.net.URL;
+
+import static org.junit.Assert.assertEquals;
+
+/** Tests for {@link KafkaSerializerWrapper}. */
+public class KafkaSerializerWrapperTest {
+ @Test
+ public void testUserCodeClassLoaderIsUsed() throws Exception {
+ final KafkaSerializerWrapperCaptureForTest wrapper =
+ new KafkaSerializerWrapperCaptureForTest();
+ final ClassLoader classLoader =
+ FlinkUserCodeClassLoaders.childFirst(
+ new URL[0],
+ getClass().getClassLoader(),
+ new String[0],
+ throwable -> {},
+ true);
+ wrapper.open(
+ new SerializationSchema.InitializationContext() {
+ @Override
+ public MetricGroup getMetricGroup() {
+ return new UnregisteredMetricsGroup();
+ }
+
+ @Override
+ public UserCodeClassLoader getUserCodeClassLoader() {
+ return SimpleUserCodeClassLoader.create(classLoader);
+ }
+ });
+
+ assertEquals(classLoader, wrapper.getClassLoaderUsed());
+ }
Review Comment:
Could you please add another test case for previous classloader? With a descriptive test similar to yours, e.g, `testDefaultClassLoaderIsUsed`.
In this case you can use the `DummyInitializationContext`.
The test `KafkaRecordDeserializationSchemaTest` already tests this but the assertion is not on the classloader, so having default classloader assertion (e.g, `DummyInitializationContext.class.getClassLoader()`) make it clearer.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSerializerWrapperTest.java:
##########
@@ -0,0 +1,63 @@
+package org.apache.flink.connector.kafka.sink;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.util.FlinkUserCodeClassLoaders;
+import org.apache.flink.util.SimpleUserCodeClassLoader;
+import org.apache.flink.util.UserCodeClassLoader;
+
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.Test;
+
+import java.net.URL;
+
+import static org.junit.Assert.assertEquals;
+
+/** Tests for {@link KafkaSerializerWrapper}. */
+public class KafkaSerializerWrapperTest {
+ @Test
+ public void testUserCodeClassLoaderIsUsed() throws Exception {
+ final KafkaSerializerWrapperCaptureForTest wrapper =
+ new KafkaSerializerWrapperCaptureForTest();
+ final ClassLoader classLoader =
+ FlinkUserCodeClassLoaders.childFirst(
+ new URL[0],
+ getClass().getClassLoader(),
+ new String[0],
+ throwable -> {},
+ true);
+ wrapper.open(
+ new SerializationSchema.InitializationContext() {
+ @Override
+ public MetricGroup getMetricGroup() {
+ return new UnregisteredMetricsGroup();
+ }
+
+ @Override
+ public UserCodeClassLoader getUserCodeClassLoader() {
+ return SimpleUserCodeClassLoader.create(classLoader);
+ }
+ });
+
+ assertEquals(classLoader, wrapper.getClassLoaderUsed());
+ }
Review Comment:
Same comment for the deserialization test below
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java:
##########
@@ -55,17 +55,11 @@ class KafkaValueOnlyDeserializerWrapper<T> implements KafkaRecordDeserialization
}
@Override
- @SuppressWarnings("unchecked")
public void open(DeserializationSchema.InitializationContext context) throws Exception {
- ClassLoader userCodeClassLoader = context.getUserCodeClassLoader().asClassLoader();
+ ClassLoader userCodeClassLoader = selectClassLoader(context);
Review Comment:
Same here let's skip the method
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSerializerWrapper.java:
##########
@@ -61,17 +61,12 @@ class KafkaSerializerWrapper<IN> implements SerializationSchema<IN> {
this(serializerClass, isKey, Collections.emptyMap(), topicSelector);
}
- @SuppressWarnings("unchecked")
@Override
public void open(InitializationContext context) throws Exception {
- final ClassLoader userCodeClassLoader = context.getUserCodeClassLoader().asClassLoader();
+ final ClassLoader userCodeClassLoader = selectClassLoader(context);
Review Comment:
Since the `selectClassLoader` is not used in the tests, maybe we don't have to extract it into a method
--
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-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
Posted by "hugogu (via GitHub)" <gi...@apache.org>.
hugogu commented on code in PR #89:
URL: https://github.com/apache/flink-connector-kafka/pull/89#discussion_r1555389033
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSerializerWrapper.java:
##########
@@ -61,17 +61,12 @@ class KafkaSerializerWrapper<IN> implements SerializationSchema<IN> {
this(serializerClass, isKey, Collections.emptyMap(), topicSelector);
}
- @SuppressWarnings("unchecked")
@Override
public void open(InitializationContext context) throws Exception {
- final ClassLoader userCodeClassLoader = context.getUserCodeClassLoader().asClassLoader();
+ final ClassLoader userCodeClassLoader = selectClassLoader(context);
Review Comment:
Sure, removed.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaValueOnlyDeserializerWrapper.java:
##########
@@ -55,17 +55,11 @@ class KafkaValueOnlyDeserializerWrapper<T> implements KafkaRecordDeserialization
}
@Override
- @SuppressWarnings("unchecked")
public void open(DeserializationSchema.InitializationContext context) throws Exception {
- ClassLoader userCodeClassLoader = context.getUserCodeClassLoader().asClassLoader();
+ ClassLoader userCodeClassLoader = selectClassLoader(context);
Review Comment:
done
--
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-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
Posted by "hugogu (via GitHub)" <gi...@apache.org>.
hugogu commented on PR #89:
URL: https://github.com/apache/flink-connector-kafka/pull/89#issuecomment-2034668045
@MartijnVisser True. I added a couple of tests that would fail without my change. Hope it looks good now. Please let me know if anything else is 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
Re: [PR] [FLINK-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
Posted by "morazow (via GitHub)" <gi...@apache.org>.
morazow commented on PR #89:
URL: https://github.com/apache/flink-connector-kafka/pull/89#issuecomment-2041992850
Thanks @hugogu for the PR! I have added some comments, please have a look.
Do you know why the CI is failing?
--
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-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
Posted by "hugogu (via GitHub)" <gi...@apache.org>.
hugogu commented on code in PR #89:
URL: https://github.com/apache/flink-connector-kafka/pull/89#discussion_r1555387996
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSerializerWrapperTest.java:
##########
@@ -0,0 +1,63 @@
+package org.apache.flink.connector.kafka.sink;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.util.FlinkUserCodeClassLoaders;
+import org.apache.flink.util.SimpleUserCodeClassLoader;
+import org.apache.flink.util.UserCodeClassLoader;
+
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.Test;
+
+import java.net.URL;
+
+import static org.junit.Assert.assertEquals;
+
+/** Tests for {@link KafkaSerializerWrapper}. */
+public class KafkaSerializerWrapperTest {
+ @Test
+ public void testUserCodeClassLoaderIsUsed() throws Exception {
+ final KafkaSerializerWrapperCaptureForTest wrapper =
+ new KafkaSerializerWrapperCaptureForTest();
+ final ClassLoader classLoader =
+ FlinkUserCodeClassLoaders.childFirst(
+ new URL[0],
+ getClass().getClassLoader(),
+ new String[0],
+ throwable -> {},
+ true);
+ wrapper.open(
+ new SerializationSchema.InitializationContext() {
+ @Override
+ public MetricGroup getMetricGroup() {
+ return new UnregisteredMetricsGroup();
+ }
+
+ @Override
+ public UserCodeClassLoader getUserCodeClassLoader() {
+ return SimpleUserCodeClassLoader.create(classLoader);
+ }
+ });
+
+ assertEquals(classLoader, wrapper.getClassLoaderUsed());
+ }
Review Comment:
Sure. Just added a couple tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
Posted by "hugogu (via GitHub)" <gi...@apache.org>.
hugogu commented on PR #89:
URL: https://github.com/apache/flink-connector-kafka/pull/89#issuecomment-2039767446
@MartijnVisser I also fix a few style errors, please let me know if it is good now or any other change to make. Thanks for your 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
Re: [PR] [FLINK-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #89:
URL: https://github.com/apache/flink-connector-kafka/pull/89#issuecomment-2034157470
Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
Posted by "hugogu (via GitHub)" <gi...@apache.org>.
hugogu commented on code in PR #89:
URL: https://github.com/apache/flink-connector-kafka/pull/89#discussion_r1550913873
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSerializerWrapperTest.java:
##########
@@ -0,0 +1,28 @@
+package org.apache.flink.connector.kafka.sink;
+
+import org.apache.flink.streaming.connectors.kafka.testutils.SerializationTestBase;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
Review Comment:
Thanks for pointing it out. I have just updated it to get rid of Mockito.
--
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-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
Posted by "hugogu (via GitHub)" <gi...@apache.org>.
hugogu commented on PR #89:
URL: https://github.com/apache/flink-connector-kafka/pull/89#issuecomment-2034714493
@MartijnVisser May I have your advice on the use of `TemporaryClassLoaderContext` in the implementation? It looks unnecessary to me. Do you think I should just remove it in this PR?
```java
public void open(InitializationContext context) throws Exception {
final ClassLoader userCodeClassLoader = context.getUserCodeClassLoader().asClassLoader();
try (TemporaryClassLoaderContext ignored =
TemporaryClassLoaderContext.of(userCodeClassLoader)) {
serializer =
InstantiationUtil.instantiate(
serializerClass.getName(),
Serializer.class,
userCodeClassLoader);
if (serializer instanceof Configurable) {
((Configurable) serializer).configure(config);
} else {
serializer.configure(config, isKey);
}
} catch (Exception e) {
throw new IOException("Failed to instantiate the serializer of class " + serializer, e);
}
}
```
--
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-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
Posted by "hugogu (via GitHub)" <gi...@apache.org>.
hugogu commented on PR #89:
URL: https://github.com/apache/flink-connector-kafka/pull/89#issuecomment-2065806444
Hi @uce and @rmetzger, do you mind reviewing the change and let me know if any change is required before merging. Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34996][Connectors/Kafka] Use UserCodeCL to instantiate Deserializer [flink-connector-kafka]
Posted by "hugogu (via GitHub)" <gi...@apache.org>.
hugogu commented on PR #89:
URL: https://github.com/apache/flink-connector-kafka/pull/89#issuecomment-2049754379
@morazow Thanks for your approval. I have just rebased this PR to include the build fix made in main branch. Hopefully the build would success this time.
I also noticed that v3.1 branch, may I know if this fix can be included? Shall I raise another PR for that? Not quite sure about the release procedure yet.
cc @MartijnVisser
--
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