fix: fixes npe in timestamp encoding (#12)

* fix: fix npe in timestamp encoding

Handle the case when the timestamp written / read is null.

* refactor: remove read/write null from ts encoding

Removes the calls to read/write null from timestamp encoding.

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/cdc/TimestampEncoding.java

@@ -36,6 +36,7 @@ public class TimestampEncoding extends CustomEncoding<Timestamp> {
         .builder()
         .record("timestamp")
         .fields()
+        .requiredBoolean("isNull")
         .requiredLong("seconds")
         .requiredInt("nanos")
         .endRecord();
@@ -45,15 +46,25 @@ public class TimestampEncoding extends CustomEncoding<Timestamp> {
   @Override
   protected void write(Object datum, Encoder out) throws IOException {
     final Timestamp timestamp = (Timestamp) datum;
-    out.writeLong(timestamp.getSeconds());
-    out.writeInt(timestamp.getNanos());
+    if (timestamp == null) {
+      out.writeBoolean(true);
+    } else {
+      out.writeBoolean(false);
+      out.writeLong(timestamp.getSeconds());
+      out.writeInt(timestamp.getNanos());
+    }
   }
 
   @Override
   protected Timestamp read(Object reuse, Decoder in) throws IOException {
-    return Timestamp.ofTimeSecondsAndNanos(
-        in.readLong(),
-        in.readInt()
-    );
+    final boolean isNull = in.readBoolean();
+    if (isNull) {
+      return null;
+    } else {
+      return Timestamp.ofTimeSecondsAndNanos(
+          in.readLong(),
+          in.readInt()
+      );
+    }
   }
 }

sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/cdc/TimestampEncodingTest.java

@@ -1,6 +1,7 @@
 package org.apache.beam.sdk.io.gcp.spanner.cdc;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 
 import com.google.cloud.Timestamp;
 import java.io.ByteArrayInputStream;
@@ -57,6 +58,21 @@ public void testWriteAndReadReuseTimestamp() throws IOException {
     assertEquals(expectedTimestamp, actualTimestamp);
   }
 
+  @Test
+  public void testWriteAndReadNullTimestamp() throws IOException {
+    final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+    final BinaryEncoder encoder = EncoderFactory.get().directBinaryEncoder(outputStream, null);
+
+    encoding.write(null, encoder);
+
+    final byte[] bytes = outputStream.toByteArray();
+    final ByteArrayInputStream inputStream = new ByteArrayInputStream(bytes);
+    final BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(inputStream, null);
+    final Timestamp actualTimestamp = encoding.read(null, decoder);
+
+    assertNull(actualTimestamp);
+  }
+
   @Test(expected = ClassCastException.class)
   public void testThrowsExceptionWhenWritingNonTimestamp() throws IOException {
     encoding.write(1L, null);