find npe

Author: stankiewicz

sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/Timer.java

@@ -126,7 +126,7 @@ public static <K> Timer<K> cleared(
    */
   public abstract @Nullable PaneInfo getPaneInfo();
 
-  public abstract @Nullable CausedByDrain causedByDrain();
+  public abstract CausedByDrain causedByDrain();
 
   @Override
   public final boolean equals(@Nullable Object other) {

sdks/java/core/src/main/java/org/apache/beam/sdk/values/WindowedValues.java

@@ -148,6 +148,7 @@ public Builder<T> setRecordOffset(@Nullable Long recordOffset) {
 
     @Override
     public Builder<T> setCausedByDrain(CausedByDrain causedByDrain) {
+      checkStateNotNull(causedByDrain, "CausedByDrain is null");
       this.causedByDrain = causedByDrain;
       return this;
     }
@@ -202,6 +203,7 @@ public PaneInfo getPaneInfo() {
 
     @Override
     public CausedByDrain causedByDrain() {
+      checkStateNotNull(causedByDrain, "CausedByDrain not set");
       return causedByDrain;
     }
 

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java

@@ -1202,7 +1202,8 @@ private <K> void processTimer(
     checkNotNull(timerBundleTracker);
     try {
       currentKey = timer.getUserKey();
-      causedByDrain = timer.causedByDrain();
+      causedByDrain = Preconditions.checkNotNull(timer.causedByDrain());
+
       // add drain
       Iterator<BoundedWindow> windowIterator =
           (Iterator<BoundedWindow>) timer.getWindows().iterator();