StreamConfig extends MapConfig { public Config getStreamProperties(String streamId) { MapConfig allProperties = getAllStreamProperties(streamId); Config samzaProperties = allProperties.subset(SAMZA_PROPERTY, false); Map<String, String> filteredStreamProperties = allProperties.entrySet().stream().filter(kv -> !samzaProperties.containsKey(kv.getKey())).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); return new MapConfig(filteredStreamProperties); } StreamConfig(Config config); Set<String> getStreamIds(); String getSystem(String streamId); String getPhysicalName(String streamId); Optional<String> getStreamMsgSerde(SystemStream systemStream); Optional<String> getStreamKeySerde(SystemStream systemStream); boolean getResetOffset(SystemStream systemStream); boolean isResetOffsetConfigured(SystemStream systemStream); Optional<String> getDefaultStreamOffset(SystemStream systemStream); boolean isDefaultStreamOffsetConfigured(SystemStream systemStream); boolean getBootstrapEnabled(SystemStream systemStream); boolean getBroadcastEnabled(SystemStream systemStream); int getPriority(SystemStream systemStream); SystemStream streamIdToSystemStream(String streamId); Set<SystemStream> getSerdeStreams(String systemName); Config getStreamProperties(String streamId); boolean getIsIntermediateStream(String streamId); boolean getDeleteCommittedMessages(String streamId); boolean getIsBounded(String streamId); static final Logger LOG; static final String SAMZA_PROPERTY; static final String SYSTEM; static final String PHYSICAL_NAME; static final String MSG_SERDE; static final String KEY_SERDE; static final String CONSUMER_RESET_OFFSET; static final String CONSUMER_OFFSET_DEFAULT; static final String BOOTSTRAP; static final String PRIORITY; static final String IS_INTERMEDIATE; static final String DELETE_COMMITTED_MESSAGES; static final String IS_BOUNDED; static final String BROADCAST; static final String STREAM_PREFIX; static final String STREAM_ID_PREFIX; static final String SYSTEM_FOR_STREAM_ID; static final String PHYSICAL_NAME_FOR_STREAM_ID; static final String IS_INTERMEDIATE_FOR_STREAM_ID; static final String DELETE_COMMITTED_MESSAGES_FOR_STREAM_ID; static final String IS_BOUNDED_FOR_STREAM_ID; static final String PRIORITY_FOR_STREAM_ID; static final String CONSUMER_OFFSET_DEFAULT_FOR_STREAM_ID; static final String BOOTSTRAP_FOR_STREAM_ID; static final String BROADCAST_FOR_STREAM_ID; }
@Test public void testGetStreamProperties() { assertEquals(new MapConfig(), new StreamConfig(new MapConfig()).getStreamProperties(STREAM_ID)); String propertyName = "stream.property.name"; StreamConfig streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.STREAM_ID_PREFIX, OTHER_STREAM_ID) + propertyName, UNUSED_VALUE))); assertEquals(new MapConfig(), streamConfig.getStreamProperties(STREAM_ID)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.STREAM_PREFIX, SYSTEM, OTHER_STREAM_ID) + propertyName, UNUSED_VALUE, String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, OTHER_STREAM_ID), SYSTEM))); assertEquals(new MapConfig(), streamConfig.getStreamProperties(STREAM_ID)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.STREAM_PREFIX, SYSTEM, STREAM_ID) + propertyName, UNUSED_VALUE))); assertEquals(new MapConfig(), streamConfig.getStreamProperties(STREAM_ID)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.STREAM_ID_PREFIX, STREAM_ID) + SAMZA_IGNORED_PROPERTY, UNUSED_VALUE))); assertEquals(new MapConfig(), streamConfig.getStreamProperties(STREAM_ID)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.STREAM_PREFIX, SYSTEM, STREAM_ID) + SAMZA_IGNORED_PROPERTY, UNUSED_VALUE, String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, STREAM_ID), SYSTEM))); assertEquals(new MapConfig(), streamConfig.getStreamProperties(STREAM_ID)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.STREAM_PREFIX, SYSTEM, STREAM_ID) + propertyName, UNUSED_VALUE, String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, STREAM_ID), SYSTEM, String.format(StreamConfig.PHYSICAL_NAME_FOR_STREAM_ID, STREAM_ID), PHYSICAL_STREAM))); assertEquals(new MapConfig(), streamConfig.getStreamProperties(PHYSICAL_STREAM)); String propertyValue = "value"; streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.STREAM_ID_PREFIX, STREAM_ID) + propertyName, propertyValue))); assertEquals(new MapConfig(ImmutableMap.of(propertyName, propertyValue)), streamConfig.getStreamProperties(STREAM_ID)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.STREAM_PREFIX, SYSTEM, STREAM_ID) + propertyName, propertyValue, String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, STREAM_ID), SYSTEM))); assertEquals(new MapConfig(ImmutableMap.of(propertyName, propertyValue)), streamConfig.getStreamProperties(STREAM_ID)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(SystemConfig.SYSTEM_DEFAULT_STREAMS_PREFIX_FORMAT, SYSTEM) + propertyName, propertyValue, String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, STREAM_ID), SYSTEM))); assertEquals(new MapConfig(ImmutableMap.of(propertyName, propertyValue)), streamConfig.getStreamProperties(STREAM_ID)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.STREAM_PREFIX, SYSTEM, PHYSICAL_STREAM) + propertyName, propertyValue, String.format(StreamConfig.STREAM_PREFIX, SYSTEM, STREAM_ID) + propertyName, UNUSED_VALUE, String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, STREAM_ID), SYSTEM, String.format(StreamConfig.PHYSICAL_NAME_FOR_STREAM_ID, STREAM_ID), PHYSICAL_STREAM))); assertEquals(new MapConfig(ImmutableMap.of(propertyName, propertyValue)), streamConfig.getStreamProperties(STREAM_ID)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.STREAM_ID_PREFIX, STREAM_ID) + propertyName, propertyValue, String.format(StreamConfig.STREAM_PREFIX, SYSTEM, STREAM_ID) + propertyName, UNUSED_VALUE, String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, STREAM_ID), SYSTEM))); assertEquals(new MapConfig(ImmutableMap.of(propertyName, propertyValue)), streamConfig.getStreamProperties(STREAM_ID)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.STREAM_PREFIX, SYSTEM, STREAM_ID) + propertyName, propertyValue, String.format(SystemConfig.SYSTEM_DEFAULT_STREAMS_PREFIX_FORMAT, SYSTEM) + propertyName, UNUSED_VALUE, String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, STREAM_ID), SYSTEM))); assertEquals(new MapConfig(ImmutableMap.of(propertyName, propertyValue)), streamConfig.getStreamProperties(STREAM_ID)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.STREAM_ID_PREFIX, STREAM_ID) + "from.stream.id.property", "fromStreamIdValue", String.format(StreamConfig.STREAM_ID_PREFIX, STREAM_ID) + "from.stream.id.other.property", "fromStreamIdOtherValue", String.format(StreamConfig.STREAM_PREFIX, SYSTEM, STREAM_ID) + "from.system.stream.property", "fromSystemStreamValue", String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, STREAM_ID), SYSTEM))); assertEquals(new MapConfig(ImmutableMap.of( "from.stream.id.property", "fromStreamIdValue", "from.stream.id.other.property", "fromStreamIdOtherValue", "from.system.stream.property", "fromSystemStreamValue")), streamConfig.getStreamProperties(STREAM_ID)); }