StreamConfig extends MapConfig { public Set<SystemStream> getSerdeStreams(String systemName) { Config subConf = subset(String.format("systems.%s.streams.", systemName, true)); Set<SystemStream> legacySystemStreams = subConf.keySet().stream() .filter(k -> k.endsWith(MSG_SERDE) || k.endsWith(KEY_SERDE)) .map(k -> { String streamName = k.substring(0, k.length() - 16 ); return new SystemStream(systemName, streamName); }) .collect(Collectors.toSet()); Set<SystemStream> systemStreams = subset(STREAMS_PREFIX).keySet().stream() .filter(k -> k.endsWith(MSG_SERDE) || k.endsWith(KEY_SERDE)) .map(k -> k.substring(0, k.length() - 16 )) .filter(streamId -> systemName.equals(getSystem(streamId))) .map(streamId -> streamIdToSystemStream(streamId)).collect(Collectors.toSet()); return Sets.union(legacySystemStreams, systemStreams).immutableCopy(); } 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 testGetSerdeStreams() { assertEquals(Collections.emptySet(), new StreamConfig(new MapConfig()).getSerdeStreams(SYSTEM)); StreamConfig streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.STREAM_ID_PREFIX, STREAM_ID) + SAMZA_IGNORED_PROPERTY, UNUSED_VALUE))); assertEquals(Collections.emptySet(), streamConfig.getSerdeStreams(SYSTEM)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.STREAM_ID_PREFIX, STREAM_ID) + StreamConfig.KEY_SERDE, UNUSED_VALUE, String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, STREAM_ID), "otherSystem"))); assertEquals(Collections.emptySet(), streamConfig.getSerdeStreams(SYSTEM)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.STREAM_PREFIX, SYSTEM, STREAM_ID) + SAMZA_IGNORED_PROPERTY, UNUSED_VALUE))); assertEquals(Collections.emptySet(), streamConfig.getSerdeStreams(SYSTEM)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.STREAM_PREFIX, "otherSystem", STREAM_ID) + StreamConfig.KEY_SERDE, UNUSED_VALUE))); assertEquals(Collections.emptySet(), streamConfig.getSerdeStreams(SYSTEM)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.STREAM_ID_PREFIX, STREAM_ID) + StreamConfig.KEY_SERDE, UNUSED_VALUE, String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, STREAM_ID), "otherSystem"))); assertEquals(Collections.emptySet(), streamConfig.getSerdeStreams(SYSTEM)); String serdeValue = "my.serde.class"; streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, STREAM_ID), SYSTEM, String.format(StreamConfig.STREAM_ID_PREFIX, STREAM_ID) + StreamConfig.KEY_SERDE, serdeValue))); assertEquals(Collections.singleton(new SystemStream(SYSTEM, STREAM_ID)), streamConfig.getSerdeStreams(SYSTEM)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, STREAM_ID), SYSTEM, String.format(StreamConfig.STREAM_ID_PREFIX, STREAM_ID) + StreamConfig.MSG_SERDE, serdeValue))); assertEquals(Collections.singleton(new SystemStream(SYSTEM, STREAM_ID)), streamConfig.getSerdeStreams(SYSTEM)); streamConfig = new StreamConfig(new MapConfig(ImmutableMap.of( String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, STREAM_ID), SYSTEM, String.format(StreamConfig.PHYSICAL_NAME_FOR_STREAM_ID, STREAM_ID), PHYSICAL_STREAM, String.format(StreamConfig.STREAM_ID_PREFIX, STREAM_ID) + StreamConfig.KEY_SERDE, serdeValue))); assertEquals(Collections.singleton(new SystemStream(SYSTEM, PHYSICAL_STREAM)), streamConfig.getSerdeStreams(SYSTEM)); streamConfig = new StreamConfig(new MapConfig( ImmutableMap.of(String.format(StreamConfig.STREAM_PREFIX, SYSTEM, STREAM_ID) + StreamConfig.KEY_SERDE, serdeValue))); assertEquals(Collections.singleton(new SystemStream(SYSTEM, STREAM_ID)), streamConfig.getSerdeStreams(SYSTEM)); streamConfig = new StreamConfig(new MapConfig( ImmutableMap.of(String.format(StreamConfig.STREAM_PREFIX, SYSTEM, STREAM_ID) + StreamConfig.MSG_SERDE, serdeValue))); assertEquals(Collections.singleton(new SystemStream(SYSTEM, STREAM_ID)), streamConfig.getSerdeStreams(SYSTEM)); String streamIdWithPhysicalName = "streamIdWithPhysicalName"; streamConfig = new StreamConfig(new MapConfig(new ImmutableMap.Builder<String, String>() .put(JobConfig.JOB_DEFAULT_SYSTEM, SYSTEM) .put(String.format(StreamConfig.STREAM_ID_PREFIX, STREAM_ID) + StreamConfig.KEY_SERDE, serdeValue) .put(String.format(StreamConfig.STREAM_ID_PREFIX, STREAM_ID) + StreamConfig.MSG_SERDE, serdeValue) .put(String.format(StreamConfig.STREAM_ID_PREFIX, streamIdWithPhysicalName) + StreamConfig.KEY_SERDE, serdeValue) .put(String.format(StreamConfig.PHYSICAL_NAME_FOR_STREAM_ID, streamIdWithPhysicalName), PHYSICAL_STREAM) .put(String.format(StreamConfig.STREAM_PREFIX, SYSTEM, OTHER_STREAM_ID) + StreamConfig.KEY_SERDE, serdeValue) .build())); assertEquals(Sets.newHashSet(new SystemStream(SYSTEM, STREAM_ID), new SystemStream(SYSTEM, PHYSICAL_STREAM), new SystemStream(SYSTEM, OTHER_STREAM_ID)), streamConfig.getSerdeStreams(SYSTEM)); }