Which is weird, it looks like the

open

method wasn’t called, so the

mapper

is still null.

are you using table api or datastream api ?

then why you are playing with

JsonRowDataSerializationSchema

you can use TableDescriptor to sink data without considering schema

In https://issues.apache.org/jira/browse/FLINK-28621 we added logic to the open() method to several schema methods. Any code that was using the schemas incorrectly at the time (including several places in Flink (https://issues.apache.org/jira/browse/FLINK-28807) ) need to be adjusted.

Depends a bit on what

RillKafkaSink

actually is. A

RichFunction

would use

schema.openRuntimeContextInitializationContextAdapters.serializationAdapter(getRuntimeContext())

. A `Sink`uses

InitContext#asSerializationSchemaInitializationContext

. If neither of use are the case / available, you can do something like this:

new SerializationSchema.InitializationContext() {
                        @Override
                        public MetricGroup getMetricGroup() {
                            return new UnregisteredMetricsGroup();
                        }

                        @Override
                        public UserCodeClassLoader getUserCodeClassLoader() {
                            return SimpleUserCodeClassLoader.create(
                                    RillKafkaSink.class.getClassLoader());
                        }
                    })