Flink SerializationSchema: Could not serialize row error

I have some trouble using flink's SerializationSchema.

Here is my main code :

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DeserializationSchema<Row> sourceDeserializer = new JsonRowDeserializationSchema.Builder( /*Extract TypeInformation<Row> from an avsc schema file*/ ).build();
DataStream<Row> myDataStream = env.addSource( new MyCustomSource(sourceDeserializer) ) ;
final SinkFunction<Row> sink = new MyCustomSink(new JsonRowSerializationSchema.Builder(myDataStream.getType()).build());
myDataStream.addSink(sink).name("MyCustomSink");

env.execute("MyJob");

Here is my custom Sink Function :

import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.types.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@SuppressWarnings("serial")
public class MyCustomSink implements SinkFunction<Row> {

	private static final Logger LOGGER = LoggerFactory.getLogger(MyCustomSink.class);
	private final boolean print;
	private final SerializationSchema<Row> serializationSchema;

	public MyCustomSink(final SerializationSchema<Row> serializationSchema) {
		this.serializationSchema = serializationSchema;
	}

	@Override
	public void invoke(final Row value, final Context context) throws Exception {

		try {
			LOGGER.info("MyCustomSink- invoke : [{}]", new String(serializationSchema.serialize(value)));
		}catch (Exception e){
			LOGGER.error("MyCustomSink- Error while sending data : " + e);
		}
	}
}

And here is my custom Source Function (not sure it is useful for the problem I have) :

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.io.ByteStreams;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MyCustomSource<T> extends RichSourceFunction<T> implements ResultTypeQueryable<T> {

	/** logger */
	private static final Logger LOGGER = LoggerFactory.getLogger(MyCustomSource.class);

	/** the JSON deserializer */
	private final DeserializationSchema<T> deserializationSchema;

	public MyCustomSource(final DeserializationSchema<T> deserializer) {
		this.deserializationSchema = deserializer;
	}

	@Override
	public void open(final Configuration parameters) {
		...
	}

	@Override
	public void run(final SourceContext<T> ctx) throws Exception {
		LOGGER.info("run");
		InputStream data = ...; // Retrieve the input json data
		final T row = deserializationSchema
						.deserialize(ByteStreams.toByteArray(data));
		ctx.collect(row);
		
	}

	@Override
	public void cancel() {
		...
	}

	@Override
	public TypeInformation<T> getProducedType() {
		return deserializationSchema.getProducedType();
	}
}

Now I run my code and I send some data sequentially to my pipeline :

==> 
{
    "id": "sensor1",
	"data":{
		"rotation": 250
	}
}

Here, the data is correctly printed by my sink : MyCustomSink- invoke : [{"id":"sensor1","data":{"rotation":250}}]

==> 
{
    "id": "sensor1"
}

Here, the data is correctly printed by my sink : MyCustomSink- invoke : [{"id":"sensor1","data":null}]

==> 
{
    "id": "sensor1",
	"data":{
		"rotation": 250
	}
}

Here, there is an error on serialization. The error log printed is :

MyCustomSink- Error while sending data : java.lang.RuntimeException: Could not serialize row 'sensor1,250'. Make sure that the schema matches the input.

I do not understand at all why I have this behavior. Someone have an idea ?

Notes:

• Using Flink 1.9.2

-- EDIT --

I added the CustomSource part

-- EDIT 2 --

After more investigations, it looks like this behavior is caused by the private transient ObjectNode node of the JsonRowSerializationSchema. If I understand correctly, this is used for optimization, but seems to be the cause of my problem.

Is it the normal behavior, and if it is, what would be the correct use of this class in my case ? (Else, is there any way to bypass this problem ?)

This is a JsonRowSerializationSchema bug which has been fixed in most recent Flink versions - I believe, this PR addresses the issue above.