### Question:
ArchivedExecutionGraph implements AccessExecutionGraph, Serializable { public static ArchivedExecutionGraph createFrom(ExecutionGraph executionGraph) { final int numberVertices = executionGraph.getTotalNumberOfVertices(); Map<JobVertexID, ArchivedExecutionJobVertex> archivedTasks = new HashMap<>(numberVertices); List<ArchivedExecutionJobVertex> archivedVerticesInCreationOrder = new ArrayList<>(numberVertices); for (ExecutionJobVertex task : executionGraph.getVerticesTopologically()) { ArchivedExecutionJobVertex archivedTask = task.archive(); archivedVerticesInCreationOrder.add(archivedTask); archivedTasks.put(task.getJobVertexId(), archivedTask); } final Map<String, SerializedValue<OptionalFailure<Object>>> serializedUserAccumulators = executionGraph.getAccumulatorsSerialized(); final long[] timestamps = new long[JobStatus.values().length]; for (JobStatus jobStatus : JobStatus.values()) { final int ordinal = jobStatus.ordinal(); timestamps[ordinal] = executionGraph.getStatusTimestamp(jobStatus); } return new ArchivedExecutionGraph( executionGraph.getJobID(), executionGraph.getJobName(), archivedTasks, archivedVerticesInCreationOrder, timestamps, executionGraph.getState(), executionGraph.getFailureInfo(), executionGraph.getJsonPlan(), executionGraph.getAccumulatorResultsStringified(), serializedUserAccumulators, executionGraph.getArchivedExecutionConfig(), executionGraph.isStoppable(), executionGraph.getCheckpointCoordinatorConfiguration(), executionGraph.getCheckpointStatsSnapshot()); } ArchivedExecutionGraph(
			JobID jobID,
			String jobName,
			Map<JobVertexID, ArchivedExecutionJobVertex> tasks,
			List<ArchivedExecutionJobVertex> verticesInCreationOrder,
			long[] stateTimestamps,
			JobStatus state,
			@Nullable ErrorInfo failureCause,
			String jsonPlan,
			StringifiedAccumulatorResult[] archivedUserAccumulators,
			Map<String, SerializedValue<OptionalFailure<Object>>> serializedUserAccumulators,
			ArchivedExecutionConfig executionConfig,
			boolean isStoppable,
			@Nullable CheckpointCoordinatorConfiguration jobCheckpointingConfiguration,
			@Nullable CheckpointStatsSnapshot checkpointStatsSnapshot); @Override String getJsonPlan(); @Override JobID getJobID(); @Override String getJobName(); @Override JobStatus getState(); @Nullable @Override ErrorInfo getFailureInfo(); @Override ArchivedExecutionJobVertex getJobVertex(JobVertexID id); @Override Map<JobVertexID, AccessExecutionJobVertex> getAllVertices(); @Override Iterable<ArchivedExecutionJobVertex> getVerticesTopologically(); @Override Iterable<ArchivedExecutionVertex> getAllExecutionVertices(); @Override long getStatusTimestamp(JobStatus status); @Override CheckpointCoordinatorConfiguration getCheckpointCoordinatorConfiguration(); @Override CheckpointStatsSnapshot getCheckpointStatsSnapshot(); @Override boolean isArchived(); @Override ArchivedExecutionConfig getArchivedExecutionConfig(); @Override boolean isStoppable(); @Override StringifiedAccumulatorResult[] getAccumulatorResultsStringified(); @Override Map<String, SerializedValue<OptionalFailure<Object>>> getAccumulatorsSerialized(); static ArchivedExecutionGraph createFrom(ExecutionGraph executionGraph);  }### Answer:
@Test public void testArchive() throws IOException, ClassNotFoundException { ArchivedExecutionGraph archivedGraph = ArchivedExecutionGraph.createFrom(runtimeGraph); compareExecutionGraph(runtimeGraph, archivedGraph); }
@Test public void testSerialization() throws IOException, ClassNotFoundException { ArchivedExecutionGraph archivedGraph = ArchivedExecutionGraph.createFrom(runtimeGraph); verifySerializability(archivedGraph); }