[SAMZA-1895] Repeated definition of StreamConfig configuration prefix and keys should not be allowed

• Type: Improvement
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Labels: newbie
• Assignee: Unassigned
• Reporter: Yi Pan
• Created: 20/Sep/18 22:58
• Updated: 20/Sep/18 22:58

Description

Git grep result:

yipan-mn1:samza_master yipan$ git grep 'streams\.%s\.' | grep -v 'samza-api'
samza-aws/src/main/java/org/apache/samza/system/kinesis/KinesisConfig.java: private static final String CONFIG_STREAM_REGION = "systems.%s.streams.%s.aws.region";
samza-aws/src/main/java/org/apache/samza/system/kinesis/KinesisConfig.java: private static final String CONFIG_STREAM_ACCESS_KEY = "systems.%s.streams.%s.aws.accessKey";
samza-aws/src/main/java/org/apache/samza/system/kinesis/KinesisConfig.java: private static final String CONFIG_STREAM_SECRET_KEY = "sensitive.systems.%s.streams.%s.aws.secretKey";
samza-aws/src/main/java/org/apache/samza/system/kinesis/KinesisConfig.java: private static final String CONFIG_STREAM_KINESIS_CLIENT_LIB_CONFIG = "systems.%s.streams.%s.aws.kcl.";
samza-aws/src/test/java/org/apache/samza/system/kinesis/TestKinesisConfig.java: String ssConfigPrefix = String.format("systems.%s.streams.%s.", system, stream);
samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubConfig.java: public static final String CONFIG_STREAM_NAMESPACE = "streams.%s.eventhubs.namespace";
samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubConfig.java: public static final String CONFIG_STREAM_ENTITYPATH = "streams.%s.eventhubs.entitypath";
samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubConfig.java: public static final String CONFIG_STREAM_SAS_KEY_NAME = Config.SENSITIVE_PREFIX + "streams.%s.eventhubs.sas.keyname";
samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubConfig.java: public static final String CONFIG_STREAM_SAS_TOKEN = Config.SENSITIVE_PREFIX + "streams.%s.eventhubs.sas.token";
samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubConfig.java: public static final String CONFIG_STREAM_CONSUMER_GROUP = "streams.%s.eventhubs.consumer.group";
samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubConfig.java: public static final String CONFIG_SYSTEM_NUM_CLIENT_THREADS = "streams.%s.eventhubs.numClientThreads";
samza-core/src/main/scala/org/apache/samza/checkpoint/CheckpointTool.scala: val SSP_PATTERN = "tasknames.%s.systems.%s.streams.%s.partitions.%d"
samza-core/src/main/scala/org/apache/samza/config/StreamConfig.scala: val STREAM_PREFIX = "systems.%s.streams.%s."
samza-core/src/test/java/org/apache/samza/config/TestStreamConfig.java: private static final String SYSTEM_STREAM_PATTERN = "systems.%s.streams.%s.";
samza-core/src/test/java/org/apache/samza/config/TestStreamConfig.java: private static final String STREAM_ID_PATTERN = "streams.%s.";
samza-core/src/test/java/org/apache/samza/execution/TestJobNodeConfigurationGenerator.java: String keySerde = jobConfig.get(String.format("streams.%s.samza.key.serde", broadcastInputDesriptor.getStreamId()));
samza-core/src/test/java/org/apache/samza/execution/TestJobNodeConfigurationGenerator.java: String msgSerde = jobConfig.get(String.format("streams.%s.samza.msg.serde", broadcastInputDesriptor.getStreamId()));
samza-core/src/test/java/org/apache/samza/execution/TestJobNodeConfigurationGenerator.java: String streamCfgToOverride = String.format("streams.%s.samza.system", intermediateInputDescriptor.getStreamId());
samza-core/src/test/java/org/apache/samza/execution/TestJobNodeConfigurationGenerator.java: String streamCfgToOverride = String.format("streams.%s.samza.system", intermediateInputDescriptor.getStreamId());
samza-core/src/test/java/org/apache/samza/execution/TestJobNodeConfigurationGenerator.java: Config streamConfig = config.subset(String.format("streams.%s.samza.", streamId));
samza-core/src/test/java/org/apache/samza/execution/TestJobNodeConfigurationGenerator.java: Config intStreamConfig = config.subset(String.format("streams.%s.", streamId), true);
samza-core/src/test/java/org/apache/samza/util/TestStreamUtil.java: final String sysStreamPrefix = String.format("systems.%s.streams.%s.", TEST_SYSTEM, TEST_PHYSICAL_NAME);
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaInputDescriptor.java: private static final String CONSUMER_AUTO_OFFSET_RESET_CONFIG_KEY = "systems.%s.streams.%s.consumer.auto.offset.reset";
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaInputDescriptor.java: private static final String CONSUMER_FETCH_MESSAGE_MAX_BYTES_CONFIG_KEY = "systems.%s.streams.%s.consumer.fetch.message.max.bytes";
samza-sql/src/main/java/org/apache/samza/sql/runner/SamzaSqlApplicationRunner.java: public static final String CFG_FMT_SAMZA_STREAM_SYSTEM = "streams.%s.samza.system";
samza-test/src/main/java/org/apache/samza/test/framework/stream/CollectionStream.java: private static final String STREAM_TO_SYSTEM = "streams.%s.samza.system";
samza-test/src/main/java/org/apache/samza/test/framework/stream/CollectionStream.java: private static final String PHYSICAL_NAME = "streams.%s.samza.physical.name";
samza-tools/src/main/java/org/apache/samza/tools/benchmark/AbstractSamzaBench.java: protected static final String CFG_STREAM_SYSTEM_NAME = "streams.%s.samza.system";
samza-tools/src/main/java/org/apache/samza/tools/benchmark/AbstractSamzaBench.java: protected static final String CFG_PHYSICAL_STREAM_NAME = "streams.%s.samza.physical.name";
samza-tools/src/main/java/org/apache/samza/tools/benchmark/ConfigBasedSspGrouperFactory.java: public static final String CONFIG_STREAM_PARTITIONS = "streams.%s.partitions";

The repeated definitions should all be fold into StreamConfig.