OperatorSpec implements Serializable { public final String getOpId() { return this.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 testStreamTableJoinOperatorSpec() { StreamTableJoinFunction<String, Object, Object, TestOutputMessageEnvelope> joinFn = new TestStreamTableJoinFunction(); String tableId = "t1"; StreamTableJoinOperatorSpec<String, Object, Object, TestOutputMessageEnvelope> joinOperatorSpec = new StreamTableJoinOperatorSpec<>(tableId, joinFn, "join-3", 1, null, "3"); StreamTableJoinOperatorSpec<String, Object, Object, TestOutputMessageEnvelope> joinOpSpecCopy = (StreamTableJoinOperatorSpec<String, Object, Object, TestOutputMessageEnvelope>) OperatorSpecTestUtils.copyOpSpec(joinOperatorSpec); assertNotEquals(joinOpSpecCopy, joinOperatorSpec); assertEquals(joinOpSpecCopy.getOpId(), joinOperatorSpec.getOpId()); assertEquals(joinOpSpecCopy.getTableId(), joinOperatorSpec.getTableId()); assertArrayEquals(joinOpSpecCopy.getArgs(), joinOperatorSpec.getArgs()); }
@Test public void testSendToTableOperatorSpec() { String tableId = "t1"; SendToTableOperatorSpec<String, Integer> sendOpSpec = new SendToTableOperatorSpec<>(tableId, "output-1", 1, null, "3"); SendToTableOperatorSpec<String, Integer> sendToCopy = (SendToTableOperatorSpec<String, Integer>) OperatorSpecTestUtils .copyOpSpec(sendOpSpec); assertNotEquals(sendToCopy, sendOpSpec); assertEquals(sendToCopy.getOpId(), sendOpSpec.getOpId()); assertArrayEquals(sendToCopy.getArgs(), sendOpSpec.getArgs()); assertTrue(sendToCopy.getTableId().equals(sendOpSpec.getTableId())); }
@Test public void testBroadcastOperatorSpec() { OutputStreamImpl<TestOutputMessageEnvelope> outputStream = new OutputStreamImpl<>("output-0", new StringSerde("UTF-8"), new JsonSerdeV2<TestOutputMessageEnvelope>(), true); BroadcastOperatorSpec<TestOutputMessageEnvelope> broadcastOpSpec = new BroadcastOperatorSpec<>(outputStream, "broadcast-1"); BroadcastOperatorSpec<TestOutputMessageEnvelope> broadcastOpCopy = (BroadcastOperatorSpec<TestOutputMessageEnvelope>) OperatorSpecTestUtils .copyOpSpec(broadcastOpSpec); assertNotEquals(broadcastOpCopy, broadcastOpSpec); assertEquals(broadcastOpCopy.getOpId(), broadcastOpSpec.getOpId()); assertTrue(broadcastOpCopy.getOutputStream() != broadcastOpSpec.getOutputStream()); assertEquals(broadcastOpCopy.getOutputStream().getStreamId(), broadcastOpSpec.getOutputStream().getStreamId()); assertEquals(broadcastOpCopy.getOutputStream().isKeyed(), broadcastOpSpec.getOutputStream().isKeyed()); }