StreamEdge { StreamSpec getStreamSpec() { StreamSpec spec = (partitions == PARTITIONS_UNKNOWN) ? streamSpec : streamSpec.copyWithPartitionCount(partitions); if (isIntermediate && spec.getId().equals(spec.getPhysicalName())) { String physicalName = StreamManager.createUniqueNameForBatch(spec.getPhysicalName(), config); if (!physicalName.equals(spec.getPhysicalName())) { spec = spec.copyWithPhysicalName(physicalName); } } return spec; } StreamEdge(StreamSpec streamSpec, boolean isIntermediate, boolean isBroadcast, Config config); boolean isBroadcast(); static final int PARTITIONS_UNKNOWN; }
@Test public void testGetStreamSpec() { StreamEdge edge = new StreamEdge(spec, false, false, new MapConfig()); assertEquals(edge.getStreamSpec(), spec); assertEquals(edge.getStreamSpec().getPartitionCount(), 1 ); edge.setPartitionCount(10); assertEquals(edge.getStreamSpec().getPartitionCount(), 10); }
@Test public void testGetStreamSpec_Batch() { Map<String, String> config = new HashMap<>(); config.put(ApplicationConfig.APP_MODE, ApplicationConfig.ApplicationMode.BATCH.name()); config.put(ApplicationConfig.APP_RUN_ID, "123"); StreamSpec batchSpec = new StreamSpec("stream-1", "stream-1", "system-1"); StreamEdge edge = new StreamEdge(batchSpec, true, false, new MapConfig(config)); assertEquals(edge.getStreamSpec().getPhysicalName(), batchSpec.getPhysicalName() + "-123"); edge = new StreamEdge(spec, true, false, new MapConfig(config)); assertEquals(edge.getStreamSpec().getPhysicalName(), spec.getPhysicalName()); }