MapField having null Row value throws NullPointerException during encoding

public void testMapWithNullRowAsValue() {

Schema primitiveFieldsScema =
Schema.builder().addNullableField("color", FieldType.STRING).build();
      
FieldType valueType = FieldType.row(primitiveFieldsScema).withNullable(true);
      
Schema inputSchema =
                Schema.builder()
                      .addNullableField("mapWithValueAsRow",
                             FieldType.map(FieldType.STRING, valueType))
                      .build();

Map<String, Row> mapWithValueAsNullRow = new HashMap<>();
mapWithValueAsNullRow.put("key", null);   // Setting value of map as NULL

Row rowOfMap =  Row.withSchema(inputSchema)
                   .addValue(mapWithValueAsNullRow)
                   .build();

pipeline.apply(Create.of(rowOfMap)).setRowSchema(inputSchema);

pipeline.run().waitUntilFinish(Duration.standardMinutes(1));

}

java.lang.NullPointerException
    at org.apache.beam.sdk.coders.RowCoderGenerator$EncodeInstruction.encodeDelegate(RowCoderGenerator.java:225)
    at org.apache.beam.sdk.coders.Coder$ByteBuddy$DifqoQx1.encode(Unknown Source)
    at org.apache.beam.sdk.coders.Coder$ByteBuddy$DifqoQx1.encode(Unknown Source)
    at org.apache.beam.sdk.coders.MapCoder.encode(MapCoder.java:96)
    at org.apache.beam.sdk.coders.MapCoder.encode(MapCoder.java:69)
    at org.apache.beam.sdk.coders.MapCoder.encode(MapCoder.java:43)
    at org.apache.beam.sdk.coders.RowCoderGenerator$EncodeInstruction.encodeDelegate(RowCoderGenerator.java:234)
    at org.apache.beam.sdk.coders.Coder$ByteBuddy$9dwdE80N.encode(Unknown Source)
    at org.apache.beam.sdk.coders.Coder$ByteBuddy$9dwdE80N.encode(Unknown Source)
    at org.apache.beam.sdk.coders.RowCoder.encode(RowCoder.java:145)
    at org.apache.beam.sdk.schemas.SchemaCoder.encode(SchemaCoder.java:81)
    at org.apache.beam.sdk.coders.Coder.encode(Coder.java:136)
    at org.apache.beam.sdk.util.CoderUtils.encodeToSafeStream(CoderUtils.java:82)
    at org.apache.beam.sdk.util.CoderUtils.encodeToByteArray(CoderUtils.java:66)
    at org.apache.beam.sdk.util.CoderUtils.encodeToByteArray(CoderUtils.java:51)
    at org.apache.beam.sdk.util.CoderUtils.clone(CoderUtils.java:141)
    at org.apache.beam.sdk.util.MutationDetectors$CodedValueMutationDetector.<init>(MutationDetectors.java:113)
    at org.apache.beam.sdk.util.MutationDetectors.forValueWithCoder(MutationDetectors.java:44)
    at org.apache.beam.runners.direct.ImmutabilityCheckingBundleFactory$ImmutabilityEnforcingBundle.add(ImmutabilityCheckingBundleFactory.java:112)
    at org.apache.beam.runners.direct.BoundedReadEvaluatorFactory$BoundedReadEvaluator.processElement(BoundedReadEvaluatorFactory.java:151)
    at org.apache.beam.runners.direct.DirectTransformExecutor.processElements(DirectTransformExecutor.java:160)
    at org.apache.beam.runners.direct.DirectTransformExecutor.run(DirectTransformExecutor.java:124)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)