[
https://issues.apache.org/jira/browse/GOBBLIN-1915?focusedWorklogId=881252&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-881252
]
ASF GitHub Bot logged work on GOBBLIN-1915:
-------------------------------------------
Author: ASF GitHub Bot
Created on: 21/Sep/23 21:02
Start Date: 21/Sep/23 21:02
Worklog Time Spent: 10m
Work Description: phet commented on code in PR #3785:
URL: https://github.com/apache/gobblin/pull/3785#discussion_r1333586486
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/GobblinTemporalConfigurationKeys.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.temporal;
+
+import org.apache.gobblin.annotation.Alpha;
+import org.apache.gobblin.temporal.cluster.NestingExecWorker;
+
+
+/**
+ * A central place for configuration related constants of a Gobblin Temporal.
+ */
+@Alpha
+public interface GobblinTemporalConfigurationKeys {
+
+ String PREFIX = "gobblin.temporal.";
+
+ String WORKER_CLASS = PREFIX + "worker.class";
+ String DEFAULT_WORKER_CLASS = NestingExecWorker.class.getName();
Review Comment:
this worker with its workflow that I originally wrote for load testing
temporal's capabilities is fairly complicated (at least it's seemed that way
when I've described it to others).
to merely demonstrate how a worker might look, would it make better sense to
choose as a default one, who implements more of a toy, "hello world" workflow
of one activity?
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/GobblinTemporalConfigurationKeys.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.temporal;
+
+import org.apache.gobblin.annotation.Alpha;
+import org.apache.gobblin.temporal.cluster.NestingExecWorker;
+
+
+/**
+ * A central place for configuration related constants of a Gobblin Temporal.
+ */
+@Alpha
+public interface GobblinTemporalConfigurationKeys {
+
+ String PREFIX = "gobblin.temporal.";
+
+ String WORKER_CLASS = PREFIX + "worker.class";
+ String DEFAULT_WORKER_CLASS = NestingExecWorker.class.getName();
+ String GOBBLIN_TEMPORAL_NAMESPACE = PREFIX + "namespace";
+ String DEFAULT_GOBBLIN_TEMPORAL_NAMESPACE = PREFIX + "namespace";
+
+ String GOBBLIN_TEMPORAL_TASK_QUEUE = PREFIX + "task.queue.name";
+ String DEFAULT_GOBBLIN_TEMPORAL_TASK_QUEUE = "GobblinTemporalTaskQueue";
+
+ /**
+ * Number of worker processes to spin up per task runner
+ * NOTE: If this size is too large, your container can OOM and halt
execution unexpectedly. It's recommended not to touch
+ * this parameter
+ */
+ String TEMPORAL_NUM_WORKERS_PER_CONTAINER = PREFIX +
"num.workers.per.container";
+ int DEFAULT_TEMPORAL_NUM_WORKERS_PER_CONTAINERS = 1;
+
+ String TEMPORAL_TASK_SIZE = PREFIX + "task.size";
+ String TEMPORAL_TASK_MAX_BRANCHES_PER_TREE = PREFIX +
"task.maxBranchesPerTree";
+ String TEMPORAL_TASK_MAX_SUB_TREES_PER_TREE = PREFIX +
"task.maxSubTreesPerTree";
Review Comment:
if you do decide to use a different default worker, these three would all go
away
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/joblauncher/GobblinTemporalJobLauncherListener.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.temporal.joblauncher;
+
+import com.google.common.base.Optional;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.gobblin.cluster.GobblinHelixJobLauncher;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.runtime.JobContext;
+import org.apache.gobblin.runtime.JobState;
+import org.apache.gobblin.runtime.listeners.AbstractJobListener;
+import org.apache.gobblin.runtime.listeners.JobListener;
+
+
+/**
+ * A job listener used when {@link GobblinHelixJobLauncher} launches a job.
Review Comment:
`GobblinTemporalJobLauncher`?
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/joblauncher/GobblinTemporalPlanningJobLauncherMetrics.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.temporal.joblauncher;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.gobblin.cluster.HelixJobsMapping;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.instrumented.StandardMetricsBridge;
+import org.apache.gobblin.metrics.ContextAwareMeter;
+import org.apache.gobblin.metrics.ContextAwareTimer;
+import org.apache.gobblin.metrics.MetricContext;
+
+
+public class GobblinTemporalPlanningJobLauncherMetrics extends
StandardMetricsBridge.StandardMetrics {
Review Comment:
NBD to leave, for now... but wondering (since no javadoc to clarify), is
this kafka-ingestion-specific or general-purpose to any workload on temporal?
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/cluster/AbstractTemporalWorker.java:
##########
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.temporal.cluster;
+
+import com.typesafe.config.Config;
+
+import io.temporal.client.WorkflowClient;
+import io.temporal.worker.Worker;
+import io.temporal.worker.WorkerFactory;
+
+import org.apache.gobblin.temporal.GobblinTemporalConfigurationKeys;
+import org.apache.gobblin.util.ConfigUtils;
+
+
+public abstract class AbstractTemporalWorker {
Review Comment:
this class is the one part of my original load-testing that I'd recommend to
keep, even if we adopt a "toy" worker as the default
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/joblauncher/GobblinJobLauncher.java:
##########
@@ -0,0 +1,264 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.temporal.joblauncher;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigValueFactory;
+import java.io.IOException;
+import java.net.URI;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import javax.annotation.Nullable;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.annotation.Alpha;
+import org.apache.gobblin.cluster.GobblinClusterConfigurationKeys;
+import org.apache.gobblin.cluster.GobblinClusterUtils;
+import org.apache.gobblin.cluster.HelixUtils;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.metrics.Tag;
+import org.apache.gobblin.metrics.event.CountEventBuilder;
+import org.apache.gobblin.metrics.event.JobEvent;
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.rest.LauncherTypeEnum;
+import org.apache.gobblin.runtime.AbstractJobLauncher;
+import org.apache.gobblin.runtime.JobException;
+import org.apache.gobblin.runtime.JobLauncher;
+import org.apache.gobblin.runtime.JobState;
+import org.apache.gobblin.runtime.TaskStateCollectorService;
+import org.apache.gobblin.runtime.listeners.JobListener;
+import org.apache.gobblin.runtime.util.StateStores;
+import org.apache.gobblin.source.workunit.WorkUnit;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.ParallelRunner;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+/**
+ * An implementation of {@link JobLauncher} that launches a Gobblin job using
the Temporal task framework.
+ * Most of this code is lifted from {@link
org.apache.gobblin.cluster.GobblinHelixJobLauncher} and maybe in the future
+ * it may make sense to converge the code once Temporal on Gobblin is in a
mature state.
Review Comment:
I agree w/ the underlying sentiment! given there are no `io.temporal`
imports, however, I wonder whether the statement belongs better in the
`GobblinTemporalJobLauncher` javadoc
Issue Time Tracking
-------------------
Worklog Id: (was: 881252)
Time Spent: 20m (was: 10m)
> Create a module that depends on Temporal instead of Helix Task Framework
> ------------------------------------------------------------------------
>
> Key: GOBBLIN-1915
> URL: https://issues.apache.org/jira/browse/GOBBLIN-1915
> Project: Apache Gobblin
> Issue Type: Improvement
> Reporter: Matthew Ho
> Priority: Major
> Time Spent: 20m
> Remaining Estimate: 0h
>
> Helix task framework is slowly fading away as a technology. The industry has
> adopted Temporal as a reliable task framework with good observability, low
> boiler plate, and a theoretical higher throughput. In order to explore this
> technology further, we should create a separate module that is able to spin
> up yarn app master that leverages Temporal instead of Helix TF
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
