StreamEdge { Config generateConfig() { Map<String, String> streamConfig = new HashMap<>(); StreamSpec spec = getStreamSpec(); String streamId = spec.getId(); streamConfig.put(String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, streamId), spec.getSystemName()); streamConfig.put(String.format(StreamConfig.PHYSICAL_NAME_FOR_STREAM_ID, streamId), spec.getPhysicalName()); if (isIntermediate()) { streamConfig.put(String.format(StreamConfig.IS_INTERMEDIATE_FOR_STREAM_ID, streamId), "true"); streamConfig.put(String.format(StreamConfig.DELETE_COMMITTED_MESSAGES_FOR_STREAM_ID, streamId), "true"); if (ApplicationConfig.ApplicationMode.BATCH.equals(new ApplicationConfig(config).getAppMode())) { streamConfig.put(String.format(StreamConfig.CONSUMER_OFFSET_DEFAULT_FOR_STREAM_ID, streamId), "oldest"); } streamConfig.put(String.format(StreamConfig.PRIORITY_FOR_STREAM_ID, streamId), String.valueOf(Integer.MAX_VALUE)); } spec.getConfig().forEach((property, value) -> { streamConfig.put(String.format(StreamConfig.STREAM_ID_PREFIX, streamId) + property, value); }); return new MapConfig(streamConfig); } StreamEdge(StreamSpec streamSpec, boolean isIntermediate, boolean isBroadcast, Config config); boolean isBroadcast(); static final int PARTITIONS_UNKNOWN; }
@Test public void testGenerateConfig() { StreamSpec spec = new StreamSpec("stream-1", "physical-stream-1", "system-1", Collections.singletonMap("property1", "haha")); StreamEdge edge = new StreamEdge(spec, false, false, new MapConfig()); Config config = edge.generateConfig(); StreamConfig streamConfig = new StreamConfig(config); assertEquals(streamConfig.getSystem(spec.getId()), "system-1"); assertEquals(streamConfig.getPhysicalName(spec.getId()), "physical-stream-1"); assertEquals(streamConfig.getIsIntermediateStream(spec.getId()), false); assertEquals(streamConfig.getStreamProperties(spec.getId()).get("property1"), "haha"); spec = new StreamSpec("stream-1", "physical-stream-1", "system-1", Collections.singletonMap("property1", "haha")); edge = new StreamEdge(spec, false, false, new MapConfig()); config = edge.generateConfig(); streamConfig = new StreamConfig(config); edge = new StreamEdge(spec, true, false, new MapConfig()); config = edge.generateConfig(); streamConfig = new StreamConfig(config); assertEquals(streamConfig.getIsIntermediateStream(spec.getId()), true); assertEquals(streamConfig.getPriority(spec.toSystemStream()), Integer.MAX_VALUE); }