You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by xi...@apache.org on 2017/05/09 16:27:38 UTC
samza git commit: SAMZA-1273: Make StreamConfig.getStreamIds() public
Repository: samza
Updated Branches:
refs/heads/master 634e568cd -> d89d1ba3a
SAMZA-1273: Make StreamConfig.getStreamIds() public
Making StreamConfig.getStreamIds() public so config provider can scan through all the configured streams and expand some properties if needed.
Author: Xinyu Liu <xi...@xiliu-ld.linkedin.biz>
Reviewers: Jake Maes <jm...@apache.org>
Closes #172 from xinyuiscool/SAMZA-1273
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/d89d1ba3
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/d89d1ba3
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/d89d1ba3
Branch: refs/heads/master
Commit: d89d1ba3a28c89fde34c87a989367328c682d03e
Parents: 634e568
Author: Xinyu Liu <xi...@xiliu-ld.linkedin.biz>
Authored: Tue May 9 09:27:28 2017 -0700
Committer: Xinyu Liu <xi...@xiliu-ld.linkedin.biz>
Committed: Tue May 9 09:27:28 2017 -0700
----------------------------------------------------------------------
.../scala/org/apache/samza/config/StreamConfig.scala | 14 +++++++++-----
1 file changed, 9 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/d89d1ba3/samza-core/src/main/scala/org/apache/samza/config/StreamConfig.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/config/StreamConfig.scala b/samza-core/src/main/scala/org/apache/samza/config/StreamConfig.scala
index c490642..389a883 100644
--- a/samza-core/src/main/scala/org/apache/samza/config/StreamConfig.scala
+++ b/samza-core/src/main/scala/org/apache/samza/config/StreamConfig.scala
@@ -152,6 +152,15 @@ class StreamConfig(config: Config) extends ScalaMapConfig(config) with Logging {
}
/**
+ * Gets the stream IDs of all the streams defined in the config
+ * @return collection of stream IDs
+ */
+ def getStreamIds(): Iterable[String] = {
+ // StreamIds are not allowed to have '.' so the first index of '.' marks the end of the streamId.
+ subset(StreamConfig.STREAMS_PREFIX).asScala.keys.map(key => key.substring(0, key.indexOf(".")))
+ }
+
+ /**
* Gets the specified Samza property for a SystemStream. A Samza property is a property that controls how Samza
* interacts with the stream, as opposed to a property of the stream itself.
*
@@ -246,11 +255,6 @@ class StreamConfig(config: Config) extends ScalaMapConfig(config) with Logging {
new MapConfig(java.util.Arrays.asList(inheritedLegacyProperties, allProperties))
}
- private def getStreamIds(): Iterable[String] = {
- // StreamIds are not allowed to have '.' so the first index of '.' marks the end of the streamId.
- subset(StreamConfig.STREAMS_PREFIX).asScala.keys.map(key => key.substring(0, key.indexOf(".")))
- }
-
private def getStreamIdsForSystem(system: String): Iterable[String] = {
getStreamIds().filter(streamId => system.equals(getSystem(streamId)))
}