reswqa commented on code in PR #24182:
URL: https://github.com/apache/flink/pull/24182#discussion_r1464593996
##########
flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java:
##########
@@ -255,6 +256,51 @@ public class PipelineOptions {
+ " sure that only tags
are written.")
.build());
+ public static final ConfigOption<Map<String, String>> SERIALIZATION_CONFIG
=
Review Comment:
Could you please post a screenshot corresponding to the rendered config
option.
##########
flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfig.java:
##########
@@ -69,8 +75,11 @@ public final class SerializerConfig implements Serializable {
private LinkedHashSet<Class<?>> registeredPojoTypes = new
LinkedHashSet<>();
- private LinkedHashMap<Class<?>, Class<TypeInfoFactory<?>>>
registeredTypeFactories =
- new LinkedHashMap<>();
+ // Order is not required as we will traverse the type hierarchy up to find
the closest type
+ // information factory
+ // when extracting the type information.
Review Comment:
This new line seems unnecessary.
##########
flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfig.java:
##########
@@ -408,4 +431,104 @@ private <T extends Class> T loadClass(
throw new IllegalArgumentException(errorMessage, e);
}
}
+
+ private void parseSerializationConfigWithExceptionHandling(
+ ClassLoader classLoader, Map<String, String> serializationConfigs)
{
+ try {
+ parseSerializationConfig(classLoader, serializationConfigs);
+ } catch (Exception e) {
+ throw new IllegalArgumentException(
+ String.format("Could not configure serializers from %s.",
serializationConfigs),
+ e);
+ }
+ }
+
+ private void parseSerializationConfig(
+ ClassLoader classLoader, Map<String, String> serializationConfigs)
{
+ final Map<Class<?>, Map<String, String>> serializationConfigByClass =
+ serializationConfigs.entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ e ->
+ loadClass(
+ e.getKey(),
+ classLoader,
+ "Could not load class
for serialization config"),
+ e ->
ConfigurationUtils.parseStringToMap(e.getValue())));
+ for (Map.Entry<Class<?>, Map<String, String>> entry :
+ serializationConfigByClass.entrySet()) {
+ Class<?> type = entry.getKey();
+ Map<String, String> config = entry.getValue();
+ String configType = config.get("type");
Review Comment:
We may need to throw more explicit exceptions for Null 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-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
