JobGraphJsonGenerator { @VisibleForTesting Map<String, Object> operatorToMap(OperatorSpec spec) { Map<String, Object> map = new HashMap<>(); map.put("opCode", spec.getOpCode().name()); map.put("opId", spec.getOpId()); map.put("sourceLocation", spec.getSourceLocation()); Collection<OperatorSpec> nextOperators = spec.getRegisteredOperatorSpecs(); map.put("nextOperatorIds", nextOperators.stream().map(OperatorSpec::getOpId).collect(Collectors.toSet())); if (spec instanceof OutputOperatorSpec) { OutputStreamImpl outputStream = ((OutputOperatorSpec) spec).getOutputStream(); map.put("outputStreamId", outputStream.getStreamId()); } else if (spec instanceof PartitionByOperatorSpec) { OutputStreamImpl outputStream = ((PartitionByOperatorSpec) spec).getOutputStream(); map.put("outputStreamId", outputStream.getStreamId()); } if (spec instanceof StreamTableJoinOperatorSpec) { String tableId = ((StreamTableJoinOperatorSpec) spec).getTableId(); map.put("tableId", tableId); } if (spec instanceof SendToTableOperatorSpec) { String tableId = ((SendToTableOperatorSpec) spec).getTableId(); map.put("tableId", tableId); } if (spec instanceof JoinOperatorSpec) { map.put("ttlMs", ((JoinOperatorSpec) spec).getTtlMs()); } return map; }    }
@Test public void testOperatorToMapForTable() { JobGraphJsonGenerator jsonGenerator = new JobGraphJsonGenerator(); Map<String, Object> map; SendToTableOperatorSpec<Object, Object> sendToTableOperatorSpec = OperatorSpecs.createSendToTableOperatorSpec("test-sent-to-table", "test-sent-to"); map = jsonGenerator.operatorToMap(sendToTableOperatorSpec); assertTrue(map.containsKey("tableId")); assertEquals(map.get("tableId"), "test-sent-to-table"); assertEquals(map.get("opCode"), OperatorSpec.OpCode.SEND_TO.name()); assertEquals(map.get("opId"), "test-sent-to"); StreamTableJoinOperatorSpec<String, String, String, String> streamTableJoinOperatorSpec = OperatorSpecs.createStreamTableJoinOperatorSpec("test-join-table", mock(StreamTableJoinFunction.class), "test-join"); map = jsonGenerator.operatorToMap(streamTableJoinOperatorSpec); assertTrue(map.containsKey("tableId")); assertEquals(map.get("tableId"), "test-join-table"); assertEquals(map.get("opCode"), OperatorSpec.OpCode.JOIN.name()); assertEquals(map.get("opId"), "test-join"); }