[beam] branch master updated: Fix NPE in CountingSource

[bo...@apache.org]

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

boyuanz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new a872aca  Fix NPE in CountingSource
     new fc41a9d  Merge pull request #13361 from Fix NPE in CountingSource
a872aca is described below

commit a872aca93099cf43429626baf0e526c2f5ed856b
Author: Boyuan Zhang <bo...@google.com>
AuthorDate: Mon Nov 16 20:10:29 2020 -0800

    Fix NPE in CountingSource
---
 .../java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java
index c1601e3..3fbd0e6 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java
@@ -435,7 +435,8 @@ public class CountingSource {
     }
 
     private long expectedValue() {
-      if (source.period.getMillis() == 0L) {
+      // Within the SDF unbounded wrapper, we will query the initial size before we start to read.
+      if (source.period.getMillis() == 0L || firstStarted == null) {
         return Long.MAX_VALUE;
       }
       double periodsElapsed =