Hello #C03G7LJTS2G I am actually trying to deserialize the string values which I get from kafka to Tuple values using a custom deserializer using below code:

class CustomDeserializer(DeserializationSchema):

    def deserialize(self, byte_array):
        """Deserializes a byte array into a tuple.

        Args:
            byte_array: The byte array to deserialize.

        Returns:
            A tuple containing the deserialized data.
        """

        string = byte_array.decode()
        return tuple(string.split(","))

    def get_produced_type(self):
        return (Types.STRING(), Types.STRING())

It gives an error which I am trying to solve since days but no luck:

raise Py4JJavaError(

py4j.protocol.Py4JJavaError: An error occurred while calling o25.fromSource.

: java.lang.NullPointerException

at org.apache.flink.connector.kafka.source.reader.deserializer.KafkaValueOnlyDeserializationSchemaWrapper.getProducedType(KafkaValueOnlyDeserializationSchemaWrapper.java:56)

at org.apache.flink.connector.kafka.source.KafkaSource.getProducedType(KafkaSource.java:216)

at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getTypeInfo(StreamExecutionEnvironment.java:2643)

at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.fromSource(StreamExecutionEnvironment.java:2015)

at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)

at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)

at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)

at java.base/java.lang.reflect.Method.invoke(Method.java:566)

at org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)

at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:374)

at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282)

at org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)

at org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)

at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)

at java.base/java.lang.Thread.run(Thread.java:834)