samza-core/src/test/java/org/apache/samza/util/TestStreamUtil.java - samza - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */
 package org.apache.samza.util;

 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
 import org.apache.samza.config.MapConfig;
 import org.apache.samza.config.StreamConfig;
 import org.apache.samza.system.StreamSpec;
 import org.junit.Test;

 import java.util.HashMap;
 import java.util.Map;

 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;


 public class TestStreamUtil {
   private static final String STREAM_ID = "t3st-Stream_Id";
   private static final String STREAM_ID_INVALID = "test#Str3amId!";

   private static final String TEST_PHYSICAL_NAME = "t3st-Physical_Name";
   private static final String TEST_PHYSICAL_NAME2 = "testPhysicalName2";
   private static final String TEST_PHYSICAL_NAME_SPECIAL_CHARS = "test://Physical.Name?";

   private static final String TEST_SYSTEM = "t3st-System_Name";
   private static final String TEST_SYSTEM_INVALID = "test:System!Name@";

   private static final String TEST_DEFAULT_SYSTEM = "testDefaultSystemName";


   // The physical name should be pulled from the StreamConfig.PHYSICAL_NAME property value.
   @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());
   }

   // The streamId should be used as the physicalName when the physical name is not specified.
   // NOTE: its either this, set to null, or exception. This seems better for backward compatibility and API brevity.
   @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());
   }

   // If the system is specified at the stream scope, use it
   @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());
   }

   // If system isn't specified at stream scope, use the default system
   @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());
   }

   // Stream scope should override default scope
   @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());
   }

   // System is required. Throw if it cannot be determined.
   @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());
   }

   // The properties in the config "streams.{streamId}.*" should be passed through to the spec.
   @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"));
   }

   // The samza properties (which are invalid for the underlying system) should be filtered out.
   @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"));
   }

   // Verify that we use a default specified with systems.x.default.stream.*, if specified
   @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", // System default property
         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")); // Uses system default
     assertEquals("5309", properties.get("segment.bytes")); // Overrides system default
   }

   // When the physicalName argument is passed explicitly it should be used, regardless of whether it is also in the config
   @Test
   public void testGetStreamPhysicalNameArgSimple() {
     Config config = buildStreamConfig(STREAM_ID,
         StreamConfig.PHYSICAL_NAME, TEST_PHYSICAL_NAME2, // This should be ignored because of the explicit arg
         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());
   }

   // Special characters are allowed for the physical name
   @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());
   }

   // Null is allowed for the physical name
   @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());
   }

   // When the system name is provided explicitly, it should be used, regardless of whether it's also in the config
   @Test
   public void testGetStreamSystemNameArgValid() {
     Config config = buildStreamConfig(STREAM_ID,
         StreamConfig.PHYSICAL_NAME, TEST_PHYSICAL_NAME, // This should be ignored because of the explicit arg
         StreamConfig.SYSTEM, TEST_SYSTEM);              // This too

     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());
   }

   // Special characters are NOT allowed for system name, because it's used as an identifier in the config.
   @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));
   }

   // Empty strings are NOT allowed for system name, because it's used as an identifier in the 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));
   }

   // Null is not allowed IllegalArgumentException system name.
   @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));
   }

   // Special characters are NOT allowed for streamId, because it's used as an identifier in the 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));
   }

   // Empty strings are NOT allowed for streamId, because it's used as an identifier in the config.
   @Test(expected = IllegalArgumentException.class)
   public void testGetStreamStreamIdEmpty() {
     Config config = buildStreamConfig("",
         StreamConfig.SYSTEM, TEST_SYSTEM);

     StreamUtil.getStreamSpec("", new StreamConfig(config));
   }

   // Null is not allowed for streamId.
   @Test(expected = IllegalArgumentException.class)
   public void testGetStreamStreamIdNull() {
     Config config = buildStreamConfig(null,
         StreamConfig.SYSTEM, TEST_SYSTEM);

     StreamUtil.getStreamSpec(null, new StreamConfig(config));
   }


   // Helper methods

   private Config buildStreamConfig(String streamId, String... kvs) {
     // inject streams.x. into each key
     for (int i = 0; i < kvs.length - 1; i += 2) {
       kvs[i] = String.format(StreamConfig.STREAM_ID_PREFIX, streamId) + kvs[i];
     }
     return buildConfig(kvs);
   }

   private Config buildConfig(String... kvs) {
     if (kvs.length % 2 != 0) {
       throw new IllegalArgumentException("There must be parity between the keys and values");
     }

     Map<String, String> configMap = new HashMap<>();
     for (int i = 0; i < kvs.length - 1; i += 2) {
       configMap.put(kvs[i], kvs[i + 1]);
     }
     return new MapConfig(configMap);
   }

   private Config addConfigs(Config original, String... kvs) {
     Map<String, String> result = new HashMap<>();
     result.putAll(original);
     result.putAll(buildConfig(kvs));
     return new MapConfig(result);
   }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/
	package org.apache.samza.util;

	import org.apache.samza.config.Config;
	import org.apache.samza.config.JobConfig;
	import org.apache.samza.config.MapConfig;
	import org.apache.samza.config.StreamConfig;
	import org.apache.samza.system.StreamSpec;
	import org.junit.Test;

	import java.util.HashMap;
	import java.util.Map;

	import static org.junit.Assert.assertEquals;
	import static org.junit.Assert.assertNull;


	public class TestStreamUtil {
	private static final String STREAM_ID = "t3st-Stream_Id";
	private static final String STREAM_ID_INVALID = "test#Str3amId!";

	private static final String TEST_PHYSICAL_NAME = "t3st-Physical_Name";
	private static final String TEST_PHYSICAL_NAME2 = "testPhysicalName2";
	private static final String TEST_PHYSICAL_NAME_SPECIAL_CHARS = "test://Physical.Name?";

	private static final String TEST_SYSTEM = "t3st-System_Name";
	private static final String TEST_SYSTEM_INVALID = "test:System!Name@";

	private static final String TEST_DEFAULT_SYSTEM = "testDefaultSystemName";


	// The physical name should be pulled from the StreamConfig.PHYSICAL_NAME property value.
	@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());
	}

	// The streamId should be used as the physicalName when the physical name is not specified.
	// NOTE: its either this, set to null, or exception. This seems better for backward compatibility and API brevity.
	@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());
	}

	// If the system is specified at the stream scope, use it
	@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());
	}

	// If system isn't specified at stream scope, use the default system
	@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());
	}

	// Stream scope should override default scope
	@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());
	}

	// System is required. Throw if it cannot be determined.
	@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());
	}

	// The properties in the config "streams.{streamId}.*" should be passed through to the spec.
	@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"));
	}

	// The samza properties (which are invalid for the underlying system) should be filtered out.
	@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"));
	}

	// Verify that we use a default specified with systems.x.default.stream.*, if specified
	@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", // System default property
	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")); // Uses system default
	assertEquals("5309", properties.get("segment.bytes")); // Overrides system default
	}

	// When the physicalName argument is passed explicitly it should be used, regardless of whether it is also in the config
	@Test
	public void testGetStreamPhysicalNameArgSimple() {
	Config config = buildStreamConfig(STREAM_ID,
	StreamConfig.PHYSICAL_NAME, TEST_PHYSICAL_NAME2, // This should be ignored because of the explicit arg
	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());
	}

	// Special characters are allowed for the physical name
	@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());
	}

	// Null is allowed for the physical name
	@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());
	}

	// When the system name is provided explicitly, it should be used, regardless of whether it's also in the config
	@Test
	public void testGetStreamSystemNameArgValid() {
	Config config = buildStreamConfig(STREAM_ID,
	StreamConfig.PHYSICAL_NAME, TEST_PHYSICAL_NAME, // This should be ignored because of the explicit arg
	StreamConfig.SYSTEM, TEST_SYSTEM); // This too

	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());
	}

	// Special characters are NOT allowed for system name, because it's used as an identifier in the config.
	@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));
	}

	// Empty strings are NOT allowed for system name, because it's used as an identifier in the 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));
	}

	// Null is not allowed IllegalArgumentException system name.
	@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));
	}

	// Special characters are NOT allowed for streamId, because it's used as an identifier in the 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));
	}

	// Empty strings are NOT allowed for streamId, because it's used as an identifier in the config.
	@Test(expected = IllegalArgumentException.class)
	public void testGetStreamStreamIdEmpty() {
	Config config = buildStreamConfig("",
	StreamConfig.SYSTEM, TEST_SYSTEM);

	StreamUtil.getStreamSpec("", new StreamConfig(config));
	}

	// Null is not allowed for streamId.
	@Test(expected = IllegalArgumentException.class)
	public void testGetStreamStreamIdNull() {
	Config config = buildStreamConfig(null,
	StreamConfig.SYSTEM, TEST_SYSTEM);

	StreamUtil.getStreamSpec(null, new StreamConfig(config));
	}


	// Helper methods

	private Config buildStreamConfig(String streamId, String... kvs) {
	// inject streams.x. into each key
	for (int i = 0; i < kvs.length - 1; i += 2) {
	kvs[i] = String.format(StreamConfig.STREAM_ID_PREFIX, streamId) + kvs[i];
	}
	return buildConfig(kvs);
	}

	private Config buildConfig(String... kvs) {
	if (kvs.length % 2 != 0) {
	throw new IllegalArgumentException("There must be parity between the keys and values");
	}

	Map<String, String> configMap = new HashMap<>();
	for (int i = 0; i < kvs.length - 1; i += 2) {
	configMap.put(kvs[i], kvs[i + 1]);
	}
	return new MapConfig(configMap);
	}

	private Config addConfigs(Config original, String... kvs) {
	Map<String, String> result = new HashMap<>();
	result.putAll(original);
	result.putAll(buildConfig(kvs));
	return new MapConfig(result);
	}
	}