OperatorSpecGraph implements Serializable { public OperatorSpecGraph clone() { if (opSpecGraphSerde == null) { throw new IllegalStateException("Cannot clone from an already deserialized OperatorSpecGraph."); } return opSpecGraphSerde.fromBytes(serializedOpSpecGraph); } OperatorSpecGraph(StreamApplicationDescriptorImpl streamAppDesc); Map<String, InputOperatorSpec> getInputOperators(); Map<String, OutputStreamImpl> getOutputStreams(); Set<String> getIntermediateBroadcastStreamIds(); Collection<OperatorSpec> getAllOperatorSpecs(); boolean hasWindowOrJoins(); OperatorSpecGraph clone();  }
@Test public void testClone() { OperatorSpecGraph operatorSpecGraph = new OperatorSpecGraph(mockAppDesc); OperatorSpecGraph clonedSpecGraph = operatorSpecGraph.clone(); OperatorSpecTestUtils.assertClonedGraph(operatorSpecGraph, clonedSpecGraph); }
@Test(expected = IOException.class) public void testCloneWithDeserializationError() throws Throwable { TestDeserializeOperatorSpec testOp = new TestDeserializeOperatorSpec(OperatorSpec.OpCode.MAP, "test-failed-op-4"); this.allOpSpecs.add(testOp); inputOpSpecMap.values().stream().findFirst().get().registerNextOperatorSpec(testOp); OperatorSpecGraph operatorSpecGraph = new OperatorSpecGraph(mockAppDesc); try { operatorSpecGraph.clone(); fail("Should have failed with serialization error"); } catch (SamzaException se) { throw se.getCause(); } }