OperatorSpec implements Serializable { final boolean isClone(OperatorSpec other) { return this != other && this.getClass().isAssignableFrom(other.getClass()) && this.opCode.equals(other.opCode) && this.opId.equals(other.opId); } OperatorSpec(OpCode opCode, String opId); void registerNextOperatorSpec(OperatorSpec<OM, ?> nextOperatorSpec); Collection<OperatorSpec<OM, ?>> getRegisteredOperatorSpecs(); final OpCode getOpCode(); final String getOpId(); final String getSourceLocation(); abstract WatermarkFunction getWatermarkFn(); abstract ScheduledFunction getScheduledFn();  }
@Test public void testInputOperatorSpec() { Serde<Object> objSerde = new Serde<Object>() { @Override public Object fromBytes(byte[] bytes) { return null; } @Override public byte[] toBytes(Object object) { return new byte[0]; } }; InputOperatorSpec inputOperatorSpec = new InputOperatorSpec( "mockStreamId", new StringSerde("UTF-8"), objSerde, null, true, "op0"); InputOperatorSpec inputOpCopy = (InputOperatorSpec) OperatorSpecTestUtils.copyOpSpec(inputOperatorSpec); assertNotEquals("Expected deserialized copy of operator spec should not be the same as the original operator spec", inputOperatorSpec, inputOpCopy); assertTrue(inputOperatorSpec.isClone(inputOpCopy)); }
@Test public void testOutputOperatorSpec() { Serde<Object> objSerde = new Serde<Object>() { @Override public Object fromBytes(byte[] bytes) { return null; } @Override public byte[] toBytes(Object object) { return new byte[0]; } }; OutputStreamImpl<KV<String, Object>> outputStrmImpl = new OutputStreamImpl<>("mockStreamId", new StringSerde("UTF-8"), objSerde, true); OutputOperatorSpec<KV<String, Object>> outputOperatorSpec = new OutputOperatorSpec<>(outputStrmImpl, "op0"); OutputOperatorSpec<KV<String, Object>> outputOpCopy = (OutputOperatorSpec<KV<String, Object>>) OperatorSpecTestUtils .copyOpSpec(outputOperatorSpec); assertNotEquals("Expected deserialized copy of operator spec should not be the same as the original operator spec", outputOperatorSpec, outputOpCopy); assertTrue(outputOperatorSpec.isClone(outputOpCopy)); }
@Test public void testSinkOperatorSpec() { SinkFunction<TestMessageEnvelope> sinkFn = (m, c, tc) -> System.out.print(m.toString()); SinkOperatorSpec<TestMessageEnvelope> sinkOpSpec = new SinkOperatorSpec<>(sinkFn, "op0"); SinkOperatorSpec<TestMessageEnvelope> sinkOpCopy = (SinkOperatorSpec<TestMessageEnvelope>) OperatorSpecTestUtils.copyOpSpec(sinkOpSpec); assertNotEquals("Expected deserialized copy of operator spec should not be the same as the original operator spec", sinkOpSpec, sinkOpCopy); assertTrue(sinkOpSpec.isClone(sinkOpCopy)); }
@Test public void testJoinOperatorSpec() { InputOperatorSpec leftOpSpec = new InputOperatorSpec( "test-input-1", new NoOpSerde<>(), new NoOpSerde<>(), null, false, "op0"); InputOperatorSpec rightOpSpec = new InputOperatorSpec( "test-input-2", new NoOpSerde<>(), new NoOpSerde<>(), null, false, "op1"); Serde<Object> objSerde = new Serde<Object>() { @Override public Object fromBytes(byte[] bytes) { return null; } @Override public byte[] toBytes(Object object) { return new byte[0]; } }; JoinFunction<String, Object, Object, TestOutputMessageEnvelope> joinFn = new TestJoinFunction(); JoinOperatorSpec<String, Object, Object, TestOutputMessageEnvelope> joinOperatorSpec = new JoinOperatorSpec<>(leftOpSpec, rightOpSpec, joinFn, new StringSerde("UTF-8"), objSerde, objSerde, 50000, "op2"); JoinOperatorSpec<String, Object, Object, TestOutputMessageEnvelope> joinOpCopy = (JoinOperatorSpec<String, Object, Object, TestOutputMessageEnvelope>) OperatorSpecTestUtils.copyOpSpec(joinOperatorSpec); assertNotEquals("Expected deserialized copy of operator spec should not be the same as the original operator spec", joinOperatorSpec, joinOpCopy); assertTrue(joinOperatorSpec.isClone(joinOpCopy)); assertTrue(joinOpCopy.getLeftInputOpSpec().isClone(leftOpSpec)); assertTrue(joinOpCopy.getRightInputOpSpec().isClone(rightOpSpec)); }
@Test public void testStreamOperatorSpecWithMapAndListInClosure() { List<Integer> integers = new ArrayList<>(1); integers.add(0, 100); List<String> keys = new ArrayList<>(1); keys.add(0, "test-1"); MapFunction<TestMessageEnvelope, TestOutputMessageEnvelope> mapFn = m -> new TestOutputMessageEnvelope(keys.get(m.getKey().hashCode() % 1), integers.get(m.getMessage().hashCode() % 1)); StreamOperatorSpec<TestMessageEnvelope, TestOutputMessageEnvelope> streamOperatorSpec = OperatorSpecs.createMapOperatorSpec(mapFn, "op0"); StreamOperatorSpec<TestMessageEnvelope, TestOutputMessageEnvelope> cloneOperatorSpec = (StreamOperatorSpec<TestMessageEnvelope, TestOutputMessageEnvelope>) OperatorSpecTestUtils.copyOpSpec(streamOperatorSpec); assertNotEquals(streamOperatorSpec, cloneOperatorSpec); assertTrue(streamOperatorSpec.isClone(cloneOperatorSpec)); MapFunction userFn = (MapFunction) Whitebox.getInternalState(streamOperatorSpec, "mapFn"); assertEquals(userFn, mapFn); assertNotEquals(streamOperatorSpec.getTransformFn(), cloneOperatorSpec.getTransformFn()); MapFunction clonedUserFn = (MapFunction) Whitebox.getInternalState(cloneOperatorSpec, "mapFn"); assertTrue(cloneOperatorSpec.getTransformFn() instanceof FlatMapFunction); assertTrue(clonedUserFn instanceof MapFunction); assertNotEquals(userFn, clonedUserFn); TestMessageEnvelope mockImsg = new TestMessageEnvelope("input-key-x", new String("value-x")); assertEquals(((MapFunction) userFn).apply(mockImsg), new TestOutputMessageEnvelope("test-1", 100)); integers.set(0, 200); keys.set(0, "test-2"); assertEquals(((MapFunction) userFn).apply(mockImsg), new TestOutputMessageEnvelope("test-2", 200)); assertEquals(((MapFunction) clonedUserFn).apply(mockImsg), new TestOutputMessageEnvelope("test-1", 100)); }
@Test public void testStreamOperatorSpecWithMapWithFunctionReference() { MapFunction<KV<String, Object>, Object> mapFn = KV::getValue; StreamOperatorSpec<KV<String, Object>, Object> streamOperatorSpec = OperatorSpecs.createMapOperatorSpec(mapFn, "op0"); StreamOperatorSpec<TestMessageEnvelope, TestOutputMessageEnvelope> cloneOperatorSpec = (StreamOperatorSpec<TestMessageEnvelope, TestOutputMessageEnvelope>) OperatorSpecTestUtils.copyOpSpec(streamOperatorSpec); assertNotEquals(streamOperatorSpec, cloneOperatorSpec); assertTrue(streamOperatorSpec.isClone(cloneOperatorSpec)); MapFunction userFn = (MapFunction) Whitebox.getInternalState(streamOperatorSpec, "mapFn"); assertEquals(userFn, mapFn); assertNotEquals(streamOperatorSpec.getTransformFn(), cloneOperatorSpec.getTransformFn()); MapFunction clonedUserFn = (MapFunction) Whitebox.getInternalState(cloneOperatorSpec, "mapFn"); assertTrue(cloneOperatorSpec.getTransformFn() instanceof FlatMapFunction); assertTrue(clonedUserFn instanceof MapFunction); assertNotEquals(userFn, clonedUserFn); }
@Test public void testStreamOperatorSpecWithMapWithEnum() { MapFunction<TestMessageEnvelope, TestOutputMessageEnvelope> mapFn = new MapWithEnum(OperatorSpecTestUtils.TestEnum.One); StreamOperatorSpec<TestMessageEnvelope, TestOutputMessageEnvelope> streamOperatorSpec = OperatorSpecs.createMapOperatorSpec(mapFn, "op0"); assertTrue(streamOperatorSpec instanceof MapOperatorSpec); StreamOperatorSpec<TestMessageEnvelope, TestOutputMessageEnvelope> cloneOperatorSpec = (StreamOperatorSpec<TestMessageEnvelope, TestOutputMessageEnvelope>) OperatorSpecTestUtils.copyOpSpec(streamOperatorSpec); assertNotEquals(streamOperatorSpec, cloneOperatorSpec); assertTrue(streamOperatorSpec.isClone(cloneOperatorSpec)); MapFunction userFn = (MapFunction) Whitebox.getInternalState(streamOperatorSpec, "mapFn"); assertEquals(userFn, mapFn); assertNotEquals(streamOperatorSpec.getTransformFn(), cloneOperatorSpec.getTransformFn()); MapFunction clonedUserFn = (MapFunction) Whitebox.getInternalState(cloneOperatorSpec, "mapFn"); assertTrue(cloneOperatorSpec.getTransformFn() instanceof FlatMapFunction); assertTrue(clonedUserFn instanceof MapWithEnum); assertNotEquals(userFn, clonedUserFn); assertTrue(((MapWithEnum) userFn).getType() == ((MapWithEnum) clonedUserFn).getType()); ((MapWithEnum) clonedUserFn).setType(OperatorSpecTestUtils.TestEnum.Two); assertTrue(((MapWithEnum) userFn).getType() != ((MapWithEnum) clonedUserFn).getType()); }