Blazer-007 commented on code in PR #4098:
URL: https://github.com/apache/gobblin/pull/4098#discussion_r1957681454
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/GobblinTemporalConfigurationKeys.java:
##########
@@ -74,4 +80,22 @@ public interface GobblinTemporalConfigurationKeys {
String DYNAMIC_SCALING_POLLING_INTERVAL_SECS = DYNAMIC_SCALING_PREFIX +
"polling.interval.seconds";
int DEFAULT_DYNAMIC_SCALING_POLLING_INTERVAL_SECS = 60;
+
+ /**
+ * Generates a unique task queue name for Gobblin Temporal task queue.
+ * The task queue name is created by concatenating a prefix from the
configuration
+ * with a randomly generated UUID.
+ *
+ * @param config the configuration object from which to retrieve the queue
prefix
+ * @return a unique task queue name
+ * @throws NullPointerException if the provided config is null
+ */
+ static String getTemporalTaskQueueName(Config config) {
+ Objects.requireNonNull(config, "Config must not be null");
+
+ final String queuePrefix =
+ ConfigUtils.getString(config,
GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_TASK_QUEUE,
+
GobblinTemporalConfigurationKeys.DEFAULT_GOBBLIN_TEMPORAL_TASK_QUEUE);
+ return queuePrefix + UUID.randomUUID();
+ }
Review Comment:
I think this impl should be written in `AbstractTemporalWorker` and not here
as this class is mainly just keys declaration
Also instead of `Objects.requireNonNull(config, "Config must not be null");`
we can directly annotate in function param.
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/GobblinTemporalConfigurationKeys.java:
##########
@@ -74,4 +80,22 @@ public interface GobblinTemporalConfigurationKeys {
String DYNAMIC_SCALING_POLLING_INTERVAL_SECS = DYNAMIC_SCALING_PREFIX +
"polling.interval.seconds";
int DEFAULT_DYNAMIC_SCALING_POLLING_INTERVAL_SECS = 60;
+
+ /**
+ * Generates a unique task queue name for Gobblin Temporal task queue.
+ * The task queue name is created by concatenating a prefix from the
configuration
+ * with a randomly generated UUID.
+ *
+ * @param config the configuration object from which to retrieve the queue
prefix
+ * @return a unique task queue name
+ * @throws NullPointerException if the provided config is null
+ */
+ static String getTemporalTaskQueueName(Config config) {
+ Objects.requireNonNull(config, "Config must not be null");
+
+ final String queuePrefix =
+ ConfigUtils.getString(config,
GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_TASK_QUEUE,
+
GobblinTemporalConfigurationKeys.DEFAULT_GOBBLIN_TEMPORAL_TASK_QUEUE);
+ return queuePrefix + UUID.randomUUID();
Review Comment:
This will create a new UUID each time when called, we need to initialize
this taskQueue only once.
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/cluster/GobblinTemporalTaskRunner.java:
##########
@@ -151,8 +151,7 @@ public GobblinTemporalTaskRunner(String applicationName,
this.containerMetrics = buildContainerMetrics();
this.numTemporalWorkers = ConfigUtils.getInt(config,
GobblinTemporalConfigurationKeys.TEMPORAL_NUM_WORKERS_PER_CONTAINER,
GobblinTemporalConfigurationKeys.DEFAULT_TEMPORAL_NUM_WORKERS_PER_CONTAINERS);
- this.temporalQueueName = ConfigUtils.getString(config,
GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_TASK_QUEUE,
- GobblinTemporalConfigurationKeys.DEFAULT_GOBBLIN_TEMPORAL_TASK_QUEUE);
+ this.temporalQueueName =
GobblinTemporalConfigurationKeys.getTemporalTaskQueueName(config);
Review Comment:
Do we even need this here ? I don't see any usage of this
--
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.
To unsubscribe, e-mail: dev-unsubscr...@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
