From 10a2e522c76b16e76fa3e5510ed2adc720d28662 Mon Sep 17 00:00:00 2001
From: Gyewon Lee <strayyyyyy@gmail.com>
Date: Mon, 9 Jul 2018 16:49:28 +0900
Subject: [PATCH] [MIST-1174] Fix NullPointerException on
 DefaultConfigDagGeneratorImpl (#1175)

This PR fixes #1174 via

* Check whether `vertexStateWithTimestamp` is null or not in `DefaultConfigDagGeneratorImpl`.

Closes #1174.
---
 .../task/DefaultConfigDagGeneratorImpl.java   | 25 +++++++++++++------
 1 file changed, 18 insertions(+), 7 deletions(-)

diff --git a/mist-core/src/main/java/edu/snu/mist/core/task/DefaultConfigDagGeneratorImpl.java b/mist-core/src/main/java/edu/snu/mist/core/task/DefaultConfigDagGeneratorImpl.java
index 10b0b356..e5274cd9 100644
--- a/mist-core/src/main/java/edu/snu/mist/core/task/DefaultConfigDagGeneratorImpl.java
+++ b/mist-core/src/main/java/edu/snu/mist/core/task/DefaultConfigDagGeneratorImpl.java
@@ -92,13 +92,24 @@ public DAG<ConfigVertex, MISTEdge> generateWithCheckpointedStates(final AvroDag
       for (final AvroVertex avroVertex : avroVertices) {
         final StateWithTimestamp vertexStateWithTimestamp = queryState.get(avroVertex.getVertexId());
         final ExecutionVertex.Type type = getVertexType(avroVertex);
-        // Create a config vertex with checkpointed states.
-        final ConfigVertex configVertex = new ConfigVertex(
-            Long.toString(configVertexId.getAndIncrement()),
-            type,
-            avroVertex.getConfiguration(),
-            vertexStateWithTimestamp.getVertexState(),
-            vertexStateWithTimestamp.getCheckpointTimestamp());
+        final ConfigVertex configVertex;
+        if (vertexStateWithTimestamp == null) {
+          // This operator is stateless.
+          // Create a config vertex without checkpointed states.
+          configVertex = new ConfigVertex(
+              Long.toString(configVertexId.getAndIncrement()),
+              type,
+              avroVertex.getConfiguration());
+        } else {
+          // This operator is stateful.
+          // Create a config vertex with checkpointed states.
+          configVertex = new ConfigVertex(
+              Long.toString(configVertexId.getAndIncrement()),
+              type,
+              avroVertex.getConfiguration(),
+              vertexStateWithTimestamp.getVertexState(),
+              vertexStateWithTimestamp.getCheckpointTimestamp());
+        }
         deserializedVertices.add(configVertex);
         configDag.addVertex(configVertex);
       }