xintongsong commented on code in PR #47:
URL: https://github.com/apache/flink-agents/pull/47#discussion_r2184196663
##########
plan/src/main/java/org/apache/flink/agents/plan/Action.java:
##########
@@ -44,7 +45,7 @@ public Action(String name, Function exec, List<Class<?
extends Event>> listenEve
this.name = name;
this.exec = exec;
this.listenEventTypes = listenEventTypes;
- exec.checkSignature(new Class[] {Event.class});
+ exec.checkSignature(new Class[] {Event.class, RunnerContext.class});
Review Comment:
This doesn't belong to this commit.
##########
api/src/main/java/org/apache/flink/agents/api/Event.java:
##########
@@ -48,4 +48,20 @@ public Object getAttr(String name) {
public void setAttr(String name, Object value) {
attributes.put(name, value);
}
+
+ /**
+ * Get the event type name. The event type name should be the fully
qualified class name of the
+ * event, whether it is a Java or Python event.
+ */
+ public String getEventType() {
+ return this.getClass().getCanonicalName();
+ }
+
+ public boolean isInputEvent() {
+ return false;
+ }
+
+ public boolean isOutputEvent() {
+ return false;
+ }
Review Comment:
Doesn't make sense to have these methods here. It introduce the chance for
subclasses to incorrectly override these methods. I'd suggest to make the type
of input/output events static constants, and compare in `EventMessage`.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/FlinkAgent.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.flink.agents.runtime;
+
+import org.apache.flink.agents.plan.WorkflowPlan;
+import org.apache.flink.agents.runtime.feedback.FeedbackKey;
+import org.apache.flink.agents.runtime.message.EventMessage;
+import
org.apache.flink.agents.runtime.message.EventMessage.EventMessageKeySelector;
+import org.apache.flink.agents.runtime.message.Message;
+import org.apache.flink.agents.runtime.operator.ActionExecutionOperatorFactory;
+import org.apache.flink.agents.runtime.operator.FeedbackOperatorFactory;
+import org.apache.flink.agents.runtime.operator.FeedbackSinkOperator;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamUtils;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.util.OutputTag;
+
+/** A utility class that bridges Flink DataStream/SQL with the Flink Agents
workflow. */
+public class FlinkAgent {
Review Comment:
1. Is this class exposed to users?
2. The name doesn't make sense. Maybe `CompileUtils`?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/python/context/PythonRunnerContextImpl.java:
##########
@@ -15,40 +15,31 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.flink.agents.runtime.context;
+package org.apache.flink.agents.runtime.python.context;
-import org.apache.flink.agents.runtime.PythonEvent;
+import org.apache.flink.agents.api.Event;
+import org.apache.flink.agents.runtime.context.RunnerContextImpl;
+import org.apache.flink.agents.runtime.python.event.PythonEvent;
import org.apache.flink.util.Preconditions;
import javax.annotation.concurrent.NotThreadSafe;
-import java.util.ArrayList;
-import java.util.List;
-
/**
* This class is used to manage the execution context when interacting with
Python functions,
* including collecting new events generated during the execution of actions.
*/
@NotThreadSafe
-public class PythonRunnerContext {
- private final List<PythonEvent> pendingEvents;
+public class PythonRunnerContextImpl extends RunnerContextImpl {
Review Comment:
This is weird. We say `RunnerContextImpl` is "for Java-based actions", and
having a `PythonRunnerContextImpl` to extend it.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/FlinkAgent.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.flink.agents.runtime;
+
+import org.apache.flink.agents.plan.WorkflowPlan;
+import org.apache.flink.agents.runtime.feedback.FeedbackKey;
+import org.apache.flink.agents.runtime.message.EventMessage;
+import
org.apache.flink.agents.runtime.message.EventMessage.EventMessageKeySelector;
+import org.apache.flink.agents.runtime.message.Message;
+import org.apache.flink.agents.runtime.operator.ActionExecutionOperatorFactory;
+import org.apache.flink.agents.runtime.operator.FeedbackOperatorFactory;
+import org.apache.flink.agents.runtime.operator.FeedbackSinkOperator;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamUtils;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.util.OutputTag;
+
+/** A utility class that bridges Flink DataStream/SQL with the Flink Agents
workflow. */
+public class FlinkAgent {
+
+ /**
+ * Connects the given DataStream to the Flink Agents workflow.
+ *
+ * @param inputDataStream The input DataStream.
+ * @param keyTypeInfo The type information of the key.
+ * @param workflowPlan The workflow plan.
+ * @param <K> The type of the key.
+ * @return The output DataStream of the workflow.
+ */
+ public static <K> DataStream<EventMessage<K>> connectToWorkflow(
+ DataStream<EventMessage<K>> inputDataStream,
Review Comment:
We should not require input datastreams to have the `EventMessage` record
type.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/FlinkAgent.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.flink.agents.runtime;
+
+import org.apache.flink.agents.plan.WorkflowPlan;
+import org.apache.flink.agents.runtime.feedback.FeedbackKey;
+import org.apache.flink.agents.runtime.message.EventMessage;
+import
org.apache.flink.agents.runtime.message.EventMessage.EventMessageKeySelector;
+import org.apache.flink.agents.runtime.message.Message;
+import org.apache.flink.agents.runtime.operator.ActionExecutionOperatorFactory;
+import org.apache.flink.agents.runtime.operator.FeedbackOperatorFactory;
+import org.apache.flink.agents.runtime.operator.FeedbackSinkOperator;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamUtils;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.util.OutputTag;
+
+/** A utility class that bridges Flink DataStream/SQL with the Flink Agents
workflow. */
+public class FlinkAgent {
+
+ /**
+ * Connects the given DataStream to the Flink Agents workflow.
+ *
+ * @param inputDataStream The input DataStream.
+ * @param keyTypeInfo The type information of the key.
+ * @param workflowPlan The workflow plan.
+ * @param <K> The type of the key.
+ * @return The output DataStream of the workflow.
+ */
+ public static <K> DataStream<EventMessage<K>> connectToWorkflow(
+ DataStream<EventMessage<K>> inputDataStream,
+ TypeInformation<K> keyTypeInfo,
+ WorkflowPlan workflowPlan) {
+ TypeInformation<EventMessage<K>> eventMessageTypeInfo =
+ inputDataStream.getTransformation().getOutputType();
+ FeedbackKey<Message> feedbackKey = new
FeedbackKey<>("feedback-pipeline", 1L);
+
+ // 1. Create feedback operator, the feedback operator is responsible
for receive event from
+ // {@code inputDataStream} and {@code feedBackSinkOperator} and
forward to action execute
+ // operator
+ DataStream<EventMessage<K>> feedbackDatastream =
+ inputDataStream
+ .keyBy(new EventMessageKeySelector(), keyTypeInfo)
Review Comment:
Shall we also consider that the input data stream might already be a keyed
stream?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
