jiayuasu commented on code in PR #3211:
URL: https://github.com/apache/sedona/pull/3211#discussion_r3698174437
##########
flink/src/main/java/org/apache/sedona/flink/SedonaContext.java:
##########
@@ -69,4 +60,40 @@ public static StreamTableEnvironment create(
func ->
tblEnv.createTemporarySystemFunction(func.getClass().getSimpleName(), func));
return tblEnv;
}
+
+ /**
+ * ExecutionConfig.registerTypeWithKryoSerializer was removed; Flink 1.19+'s
public replacement is
+ * the declarative pipeline.serialization-config option (FLIP-398), applied
here via the
+ * SerializerConfig interface's own configure(ReadableConfig, ClassLoader) —
no cast to the
+ * internal SerializerConfigImpl needed. Flink instantiates
GeometrySerde/SpatialIndexSerde itself
+ * via their no-arg constructors, so only the class names are registered
here.
+ */
+ static void registerGeometryKryoSerializers(StreamExecutionEnvironment env) {
+ List<String> kryoRegistrations =
+ Arrays.asList(
+ kryoRegistration(Point.class, GeometrySerde.class),
+ kryoRegistration(LineString.class, GeometrySerde.class),
+ kryoRegistration(Polygon.class, GeometrySerde.class),
+ kryoRegistration(MultiPoint.class, GeometrySerde.class),
+ kryoRegistration(MultiLineString.class, GeometrySerde.class),
+ kryoRegistration(MultiPolygon.class, GeometrySerde.class),
+ kryoRegistration(GeometryCollection.class, GeometrySerde.class),
+ kryoRegistration(Circle.class, GeometrySerde.class),
+ kryoRegistration(Envelope.class, GeometrySerde.class),
+ kryoRegistration(Quadtree.class, SpatialIndexSerde.class),
+ kryoRegistration(STRtree.class, SpatialIndexSerde.class));
+
+ Configuration configuration = new Configuration();
+ configuration.set(PipelineOptions.SERIALIZATION_CONFIG, kryoRegistrations);
+ env.getConfig()
+ .getSerializerConfig()
Review Comment:
`ExecutionConfig.getSerializerConfig()` is annotated `@Internal` in both
Flink 1.19 and 2.2, so this still depends on an internal API even though the
`SerializerConfigImpl` cast is gone. Please apply the configuration through the
public environment API instead:
```java
env.configure(configuration);
```
This also uses the environment’s user classloader rather than the current
thread’s context classloader. I tested this exact replacement locally against
Flink 1.19 and 2.2.1; `ModuleTest` passed all 7 tests with both versions.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
