mynameborat commented on a change in pull request #1370:
URL: https://github.com/apache/samza/pull/1370#discussion_r434083678
##########
File path:
samza-core/src/main/scala/org/apache/samza/util/CoordinatorStreamUtil.scala
##########
@@ -48,6 +48,23 @@ object CoordinatorStreamUtil extends Logging {
new MapConfig(coordinatorSystemConfig)
}
+ /**
+ * Creates coordinator stream from config.
+ *
+ * @param config to create coordinator stream.
+ */
+ def createCoordinatorStream(config: Config): Unit = {
+ val systemAdmins = new SystemAdmins(config)
+
+ // Create the coordinator stream if it doesn't exist
+ info("Creating coordinator stream")
Review comment:
This log is going repeat itself for every attempt we do this.
Can you update the log to reflect its an attempt or remove it if there is
logging within the creation flow that differentiates between actually creating
the resource vs no-op?
##########
File path:
samza-core/src/main/java/org/apache/samza/clustermanager/JobCoordinatorLaunchUtil.java
##########
@@ -59,6 +59,8 @@ public static void run(SamzaApplication app, Config config) {
}
Config fullConfig = jobConfigs.get(0);
+ // Create coordinator stream if does not exist before fetching launch
config from it.
+ CoordinatorStreamUtil.createCoordinatorStream(fullConfig);
Review comment:
Sounds inefficient and Kafka'ish. It will be good to re-evaluate this
with the metadata store work on whether metadata store abstraction needs to
have idempotent resource creation or expose a method to identify the existence
of a resource for callers. Can you create a follow up JIRA?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
