StreamUtil { public static StreamSpec getStreamSpec(String streamId, StreamConfig streamConfig) { String physicalName = streamConfig.getPhysicalName(streamId); String system = streamConfig.getSystem(streamId); Map<String, String> streamProperties = streamConfig.getStreamProperties(streamId); return new StreamSpec(streamId, physicalName, system, streamProperties); }  static SystemStream getSystemStreamFromNameOrId(Config config, String stream); static SystemStream getSystemStreamFromNames(String systemStreamName); static String getNameFromSystemStream(SystemStream systemStream); static Set<StreamSpec> getStreamSpecs(Set<String> streamIds, StreamConfig streamConfig); static StreamSpec getStreamSpec(String streamId, StreamConfig streamConfig);  }
@Test public void testGetStreamWithPhysicalNameInConfig() { Config config = buildStreamConfig(STREAM_ID, StreamConfig.PHYSICAL_NAME, TEST_PHYSICAL_NAME, StreamConfig.SYSTEM, TEST_SYSTEM); StreamSpec spec = StreamUtil.getStreamSpec(STREAM_ID, new StreamConfig(config)); assertEquals(TEST_PHYSICAL_NAME, spec.getPhysicalName()); }
@Test public void testGetStreamWithoutPhysicalNameInConfig() { Config config = buildStreamConfig(STREAM_ID, StreamConfig.SYSTEM, TEST_SYSTEM); StreamSpec spec = StreamUtil.getStreamSpec(STREAM_ID, new StreamConfig(config)); assertEquals(STREAM_ID, spec.getPhysicalName()); }
@Test public void testGetStreamWithSystemAtStreamScopeInConfig() { Config config = buildStreamConfig(STREAM_ID, StreamConfig.PHYSICAL_NAME, TEST_PHYSICAL_NAME, StreamConfig.SYSTEM, TEST_SYSTEM); StreamSpec spec = StreamUtil.getStreamSpec(STREAM_ID, new StreamConfig(config)); assertEquals(TEST_SYSTEM, spec.getSystemName()); }
@Test public void testGetStreamWithSystemAtDefaultScopeInConfig() { Config config = addConfigs(buildStreamConfig(STREAM_ID, StreamConfig.PHYSICAL_NAME, TEST_PHYSICAL_NAME), JobConfig.JOB_DEFAULT_SYSTEM, TEST_DEFAULT_SYSTEM); StreamSpec spec = StreamUtil.getStreamSpec(STREAM_ID, new StreamConfig(config)); assertEquals(TEST_DEFAULT_SYSTEM, spec.getSystemName()); }
@Test public void testGetStreamWithSystemAtBothScopesInConfig() { Config config = addConfigs(buildStreamConfig(STREAM_ID, StreamConfig.PHYSICAL_NAME, TEST_PHYSICAL_NAME, StreamConfig.SYSTEM, TEST_SYSTEM), JobConfig.JOB_DEFAULT_SYSTEM, TEST_DEFAULT_SYSTEM); StreamSpec spec = StreamUtil.getStreamSpec(STREAM_ID, new StreamConfig(config)); assertEquals(TEST_SYSTEM, spec.getSystemName()); }
@Test(expected = IllegalArgumentException.class) public void testGetStreamWithOutSystemInConfig() { Config config = buildStreamConfig(STREAM_ID, StreamConfig.PHYSICAL_NAME, TEST_PHYSICAL_NAME); StreamSpec spec = StreamUtil.getStreamSpec(STREAM_ID, new StreamConfig(config)); assertEquals(TEST_SYSTEM, spec.getSystemName()); }
@Test public void testGetStreamPropertiesPassthrough() { Config config = buildStreamConfig(STREAM_ID, StreamConfig.PHYSICAL_NAME, TEST_PHYSICAL_NAME, StreamConfig.SYSTEM, TEST_SYSTEM, "systemProperty1", "systemValue1", "systemProperty2", "systemValue2", "systemProperty3", "systemValue3"); StreamSpec spec = StreamUtil.getStreamSpec(STREAM_ID, new StreamConfig(config)); Map<String, String> properties = spec.getConfig(); assertEquals(3, properties.size()); assertEquals("systemValue1", properties.get("systemProperty1")); assertEquals("systemValue2", properties.get("systemProperty2")); assertEquals("systemValue3", properties.get("systemProperty3")); assertEquals("systemValue1", spec.get("systemProperty1")); assertEquals("systemValue2", spec.get("systemProperty2")); assertEquals("systemValue3", spec.get("systemProperty3")); }
@Test public void testGetStreamSamzaPropertiesOmitted() { Config config = buildStreamConfig(STREAM_ID, StreamConfig.PHYSICAL_NAME, TEST_PHYSICAL_NAME, StreamConfig.SYSTEM, TEST_SYSTEM, "systemProperty1", "systemValue1", "systemProperty2", "systemValue2", "systemProperty3", "systemValue3"); StreamSpec spec = StreamUtil.getStreamSpec(STREAM_ID, new StreamConfig(config)); Map<String, String> properties = spec.getConfig(); assertEquals(3, properties.size()); assertNull(properties.get(String.format(StreamConfig.PHYSICAL_NAME_FOR_STREAM_ID, STREAM_ID))); assertNull(properties.get(String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, STREAM_ID))); assertNull(spec.get(String.format(StreamConfig.PHYSICAL_NAME_FOR_STREAM_ID, STREAM_ID))); assertNull(spec.get(String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, STREAM_ID))); }
@Test public void testStreamConfigOverrides() { final String sysStreamPrefix = String.format("systems.%s.streams.%s.", TEST_SYSTEM, TEST_PHYSICAL_NAME); Config config = addConfigs(buildStreamConfig(STREAM_ID, StreamConfig.PHYSICAL_NAME, TEST_PHYSICAL_NAME, StreamConfig.SYSTEM, TEST_SYSTEM, "systemProperty1", "systemValue1", "systemProperty2", "systemValue2", "systemProperty3", "systemValue3"), sysStreamPrefix + "systemProperty4", "systemValue4", sysStreamPrefix + "systemProperty2", "systemValue8"); StreamSpec spec = StreamUtil.getStreamSpec(STREAM_ID, new StreamConfig(config)); Map<String, String> properties = spec.getConfig(); assertEquals(4, properties.size()); assertEquals("systemValue4", properties.get("systemProperty4")); assertEquals("systemValue2", properties.get("systemProperty2")); }
@Test public void testStreamConfigOverridesWithSystemDefaults() { Config config = addConfigs(buildStreamConfig(STREAM_ID, StreamConfig.PHYSICAL_NAME, TEST_PHYSICAL_NAME, StreamConfig.SYSTEM, TEST_SYSTEM, "segment.bytes", "5309"), String.format("systems.%s.default.stream.replication.factor", TEST_SYSTEM), "4", String.format("systems.%s.default.stream.segment.bytest", TEST_SYSTEM), "867" ); StreamSpec spec = StreamUtil.getStreamSpec(STREAM_ID, new StreamConfig(config)); Map<String, String> properties = spec.getConfig(); assertEquals(3, properties.size()); assertEquals("4", properties.get("replication.factor")); assertEquals("5309", properties.get("segment.bytes")); }
@Test public void testGetStreamPhysicalNameArgSimple() { Config config = buildStreamConfig(STREAM_ID, StreamConfig.PHYSICAL_NAME, TEST_PHYSICAL_NAME2, StreamConfig.SYSTEM, TEST_SYSTEM); StreamSpec spec = StreamUtil.getStreamSpec(STREAM_ID, new StreamConfig(config)); assertEquals(STREAM_ID, spec.getId()); assertEquals(TEST_PHYSICAL_NAME2, spec.getPhysicalName()); assertEquals(TEST_SYSTEM, spec.getSystemName()); }
@Test public void testGetStreamPhysicalNameArgSpecialCharacters() { Config config = buildStreamConfig(STREAM_ID, StreamConfig.PHYSICAL_NAME, TEST_PHYSICAL_NAME_SPECIAL_CHARS, StreamConfig.SYSTEM, TEST_SYSTEM); StreamSpec spec = StreamUtil.getStreamSpec(STREAM_ID, new StreamConfig(config)); assertEquals(TEST_PHYSICAL_NAME_SPECIAL_CHARS, spec.getPhysicalName()); }
@Test public void testGetStreamPhysicalNameArgNull() { Config config = buildStreamConfig(STREAM_ID, StreamConfig.PHYSICAL_NAME, null, StreamConfig.SYSTEM, TEST_SYSTEM); StreamSpec spec = StreamUtil.getStreamSpec(STREAM_ID, new StreamConfig(config)); assertNull(spec.getPhysicalName()); }
@Test public void testGetStreamSystemNameArgValid() { Config config = buildStreamConfig(STREAM_ID, StreamConfig.PHYSICAL_NAME, TEST_PHYSICAL_NAME, StreamConfig.SYSTEM, TEST_SYSTEM); StreamSpec spec = StreamUtil.getStreamSpec(STREAM_ID, new StreamConfig(config)); assertEquals(STREAM_ID, spec.getId()); assertEquals(TEST_PHYSICAL_NAME, spec.getPhysicalName()); assertEquals(TEST_SYSTEM, spec.getSystemName()); }
@Test(expected = IllegalArgumentException.class) public void testGetStreamSystemNameArgInvalid() { Config config = buildStreamConfig(STREAM_ID, StreamConfig.PHYSICAL_NAME, TEST_PHYSICAL_NAME, StreamConfig.SYSTEM, TEST_SYSTEM_INVALID); StreamUtil.getStreamSpec(STREAM_ID, new StreamConfig(config)); }
@Test(expected = IllegalArgumentException.class) public void testGetStreamSystemNameArgEmpty() { Config config = buildStreamConfig(STREAM_ID, StreamConfig.PHYSICAL_NAME, TEST_PHYSICAL_NAME, StreamConfig.SYSTEM, ""); StreamSpec spec = StreamUtil.getStreamSpec(STREAM_ID, new StreamConfig(config)); }
@Test(expected = IllegalArgumentException.class) public void testGetStreamSystemNameArgNull() { Config config = buildStreamConfig(STREAM_ID, StreamConfig.PHYSICAL_NAME, TEST_PHYSICAL_NAME, StreamConfig.SYSTEM, null); StreamUtil.getStreamSpec(STREAM_ID, new StreamConfig(config)); }
@Test(expected = IllegalArgumentException.class) public void testGetStreamStreamIdInvalid() { Config config = buildStreamConfig(STREAM_ID_INVALID, StreamConfig.SYSTEM, TEST_SYSTEM); StreamUtil.getStreamSpec(STREAM_ID_INVALID, new StreamConfig(config)); }
@Test(expected = IllegalArgumentException.class) public void testGetStreamStreamIdEmpty() { Config config = buildStreamConfig("", StreamConfig.SYSTEM, TEST_SYSTEM); StreamUtil.getStreamSpec("", new StreamConfig(config)); }
@Test(expected = IllegalArgumentException.class) public void testGetStreamStreamIdNull() { Config config = buildStreamConfig(null, StreamConfig.SYSTEM, TEST_SYSTEM); StreamUtil.getStreamSpec(null, new StreamConfig(config)); }