[MIST-1174] Fix NullPointerException on DefaultConfigDagGeneratorImpl (…

…#1175)

This PR fixes #1174 via

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

Closes #1174.

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);
       }