initial commit with high-level design for refactoring DagManager

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManagement.java

@@ -0,0 +1,32 @@
+/*
+ * 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.service.modules.orchestration;
+
+/**
+ * Responsible for defining the behavior of {@link DagTask} handling scenarios for launch, resume, kill, job start
+ * and flow completion deadlines
+ *
+ */
+public interface DagManagement {
+
+ void launchFlow();
+ void resumeFlow();
+ void killFlow();
+ void enforceFlowCompletionDeadline();
+ void enforceJobStartDeadline();
+}

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProc.java

@@ -0,0 +1,38 @@
+/*
+ * 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.service.modules.orchestration;
+
+import org.apache.gobblin.service.modules.orchestration.exception.MaybeRetryableException;
+
+
+/**
+ * Responsible to performing the actual work for a given {@link DagTask}.
+ * It processes the {@link DagTask} by first initializing its state, performing actions
+ * like updating {@link DagStateStore} and finally submiting an event to the executor.
+ * @param <S> current state of the dag node
+ * @param <R> result after processing the dag node
+ */
+public abstract class DagProc<S, R> {
+ abstract protected S initialize() throws MaybeRetryableException;
+ abstract protected R act(S state) throws MaybeRetryableException;
+ abstract protected void sendNotification(R result) throws MaybeRetryableException;
+
+ final void process() {
+ throw new UnsupportedOperationException(" Process unsupported");
+ }
+}

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagProcFactory.java

@@ -0,0 +1,29 @@
+/*
+ * 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.service.modules.orchestration;
+
+/**
+ * Factory for creating {@link DagProc} based on the visitor type {@link DagTask}.
+ */
+public interface DagProcFactory extends DagTaskVisitor<DagProc> {
+ DagProc meet(LaunchDagTask ldt);
+ DagProc meet(KillDagTask kdt);
+ DagProc meet(ResumeDagTask rdt);
+ DagProc createFor(DagTask t);
+}
+

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagTask.java

@@ -0,0 +1,32 @@
+/*
+ * 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.service.modules.orchestration;
+
+/**
+ * Defines an individual task or job in a Dag.
+ * It carries the state information required by {@link DagProc} to for its processing.
+ * Upon completion of the {@link DagProc#process()} it will mark the lease
+ * acquired by {@link org.apache.gobblin.runtime.api.MultiActiveLeaseArbiter} as complete
+ * @param <T>
+ */
+abstract class DagTask<T> {
+
+ abstract void initialize();
+ abstract void conclude();
+ abstract T host(DagTaskVisitor<T> visitor);
+}

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagTaskStream.java

@@ -0,0 +1,63 @@
+/*
+ * 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.service.modules.orchestration;
+
+import java.util.Iterator;
+
+
+/**
+ * Holds a stream of {@link DagTask} that needs to be processed by the {@link DagManager}.
+ * It provides an implementation for {@link DagManagement} defines the rules for a flow and job.
+ * Implements {@link Iterator} to provide the next {@link DagTask} if available to {@link DagManager}
+ */
+public class DagTaskStream implements Iterator<DagTask>, DagManagement {
+ @Override
+ public boolean hasNext() {
+ return false;
+ }
+
+ @Override
+ public DagTask next() {
+ return null;
+ }
+
+ @Override
+ public void launchFlow() {
+
+ }
+
+ @Override
+ public void resumeFlow() {
+
+ }
+
+ @Override
+ public void killFlow() {
+
+ }
+
+ @Override
+ public void enforceFlowCompletionDeadline() {
+
+ }
+
+ @Override
+ public void enforceJobStartDeadline() {
+
+ }
+}

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagTaskVisitor.java

@@ -0,0 +1,28 @@
+/*
+ * 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.service.modules.orchestration;
+
+/**
+ * Interface defining {@link DagTask} based on the type of visitor.
+ * @param <T>
+ */
+public interface DagTaskVisitor<T> {
+ T meet(LaunchDagTask launchDagTask);
+ T meet(KillDagTask killDagTask);
+ T meet(ResumeDagTask resumeDagTask);
+}

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/KillDagProc.java

@@ -0,0 +1,42 @@
+/*
+ * 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.service.modules.orchestration;
+
+import org.apache.gobblin.service.modules.orchestration.exception.MaybeRetryableException;
+
+
+/**
+ * An implmentation of {@link DagProc} for killing {@link DagTask}.
+ */
+public final class KillDagProc extends DagProc{
+ @Override
+ protected Object initialize() throws MaybeRetryableException {
+ return null;
+ }
+
+ @Override
+ protected Object act(Object state) throws MaybeRetryableException {
+ return null;
+ }
+
+ @Override
+ protected void sendNotification(Object result) throws MaybeRetryableException {
+
+ }
+}
+

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/KillDagTask.java

@@ -0,0 +1,37 @@
+/*
+ * 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.service.modules.orchestration;
+/**
+ * A {@link DagTask} responsible to handle kill tasks.
+ */
+public class KillDagTask extends DagTask {
+ @Override
+ void initialize() {
+
+ }
+
+ @Override
+ void conclude() {
+
+ }
+
+ @Override
+ Object host(DagTaskVisitor visitor) {
+ return null;
+ }
+}

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/LaunchDagProc.java

@@ -0,0 +1,41 @@
+/*
+ * 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.service.modules.orchestration;
+
+import org.apache.gobblin.service.modules.orchestration.exception.MaybeRetryableException;
+
+
+/**
+ * An implmentation of {@link DagProc} for launching {@link DagTask}.
+ */
+public final class LaunchDagProc extends DagProc{
+ @Override
+ protected Object initialize() throws MaybeRetryableException {
+ return null;
+ }
+
+ @Override
+ protected Object act(Object state) throws MaybeRetryableException {
+ return null;
+ }
+
+ @Override
+ protected void sendNotification(Object result) throws MaybeRetryableException {
+
+ }
+}

gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/LaunchDagTask.java

@@ -0,0 +1,37 @@
+/*
+ * 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.service.modules.orchestration;
+/**
+ * A {@link DagTask} responsible to handle launch tasks.
+ */
+public class LaunchDagTask extends DagTask {
+ @Override
+ void initialize() {
+
+ }
+
+ @Override
+ void conclude() {
+
+ }
+
+ @Override
+ Object host(DagTaskVisitor visitor) {
+ return null;
+ }
+}