[beam] branch spark-runner_structured-streaming updated: Consider null object case on RowHelpers, fixes empty side inputs tests.

[ie...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

iemejia pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/spark-runner_structured-streaming by this push:
     new 728aa1f  Consider null object case on RowHelpers, fixes empty side inputs tests.
728aa1f is described below

commit 728aa1f7d5988acfa87daad29da4fafd19eb0455
Author: Ismaël Mejía <ie...@gmail.com>
AuthorDate: Fri May 10 12:18:08 2019 +0200

    Consider null object case on RowHelpers, fixes empty side inputs tests.
---
 .../spark/structuredstreaming/translation/helpers/RowHelpers.java      | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/RowHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/RowHelpers.java
index ca88abe..da5cc96 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/RowHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/RowHelpers.java
@@ -93,6 +93,9 @@ public final class RowHelpers {
   public static <T> T extractObjectFromRow(Row value) {
     // there is only one value put in each Row by the InputPartitionReader
     byte[] bytes = (byte[]) value.get(0);
+    if (bytes == null) {
+      return null;
+    }
     ByteArrayInputStream inputStream = new ByteArrayInputStream(bytes);
     Kryo kryo = new Kryo();
     Input input = new Input(inputStream);