jiayuasu commented on code in PR #3211:
URL: https://github.com/apache/sedona/pull/3211#discussion_r3714870585
##########
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
via
+ * StreamExecutionEnvironment.configure(ReadableConfig) —
env.getConfig().getSerializerConfig() is
+ * itself @Internal in both Flink 1.19 and 2.2, so this goes through the
environment's public API
+ * instead, which also uses the environment's own user classloader rather
than the calling
+ * thread's context classloader. 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.configure(configuration);
Review Comment:
Could we keep the thread context class loader here? The one-argument
overload uses the environment class loader in PyFlink, so the current PyFlink
job fails with `ClassNotFoundException: org.locationtech.jts.geom.Point`. The
public two-argument overload is available in both Flink 1.19 and 2.2:
```suggestion
env.configure(configuration,
Thread.currentThread().getContextClassLoader());
```
--
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]
