StreamConfig extends MapConfig { public boolean getIsBounded(String streamId) { return getBoolean(String.format(IS_BOUNDED_FOR_STREAM_ID, streamId), false); } 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 testGetIsBounded() { assertFalse(new StreamConfig(new MapConfig()).getIsBounded(STREAM_ID)); StreamConfig streamConfig = new StreamConfig(new MapConfig( ImmutableMap.of(String.format(StreamConfig.IS_BOUNDED_FOR_STREAM_ID, OTHER_STREAM_ID), "true"))); assertFalse(streamConfig.getIsBounded(STREAM_ID)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.IS_BOUNDED_FOR_STREAM_ID, STREAM_ID), "true", String.format(StreamConfig.PHYSICAL_NAME_FOR_STREAM_ID, STREAM_ID), PHYSICAL_STREAM))); assertFalse(streamConfig.getIsBounded(PHYSICAL_STREAM)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.IS_BOUNDED_FOR_STREAM_ID, STREAM_ID), "true"))); assertTrue(streamConfig.getIsBounded(STREAM_ID)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.IS_BOUNDED_FOR_STREAM_ID, STREAM_ID), "false"))); assertFalse(streamConfig.getIsBounded(STREAM_ID)); }