xintongsong commented on code in PR #23: URL: https://github.com/apache/flink-agents/pull/23#discussion_r2147615019
########## api/src/main/java/org/apache/flink/agents/api/Event.java: ########## @@ -0,0 +1,65 @@ +/* + * 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.api; + +import org.apache.flink.agents.api.utils.SerializableChecker; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException; + +import java.io.Serializable; +import java.security.InvalidParameterException; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; + +/** Base class for all event types in the system. */ +public abstract class Event implements Serializable { + private static final long serialVersionUID = 1L; + private final UUID id; + private final Map<String, Object> attributes; + + public Event() { + this.id = UUID.randomUUID(); + attributes = new HashMap<>(); + } + + public UUID getId() { + return id; + } + + public Map<String, Object> getAttributes() { + return attributes; + } + + public Object getAttr(String name) { + return attributes.get(name); + } + + public void setAttr(String name, Object value) { + attributes.put(name, value); + checkSerializable(); + } + + public void checkSerializable() { + try { + SerializableChecker.checkSerializable(this); + } catch (JsonProcessingException e) { + throw new InvalidParameterException(e.getMessage()); Review Comment: Why do we need this wrapping? ########## api/src/main/java/org/apache/flink/agents/api/InputEvent.java: ########## @@ -0,0 +1,35 @@ +/* + * 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.api; + +/** Event generated by the framework, carrying an input data that arrives at the workflow. */ +public class InputEvent extends Event { + private static final long serialVersionUID = 1L; + private final Object input; + + public InputEvent(Object input) { + super(); + this.input = input; + super.checkSerializable(); Review Comment: It's hard to guarantee that all the sub-classes will call this. I think we can only do this when calling `context.sendEvent`. ########## api/pom.xml: ########## @@ -28,4 +28,17 @@ under the License. <artifactId>api</artifactId> <name>Flink Agents : API</name> + <dependencies> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-shaded-jackson</artifactId> + <version>2.14.2-16.2</version> + </dependency> + <dependency> + <groupId>org.junit.jupiter</groupId> + <artifactId>junit-jupiter</artifactId> + <version>5.10.1</version> + <scope>compile</scope> + </dependency> + </dependencies> Review Comment: 1. `junit` should be a common dependency. Should not add it for each module. 2. I'm not entirely sure whether jackson should also be a common dependency. But how do you make sure the API and Plan modules depends on the same jackson version? ########## api/src/main/java/org/apache/flink/agents/api/Event.java: ########## @@ -0,0 +1,65 @@ +/* + * 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.api; + +import org.apache.flink.agents.api.utils.SerializableChecker; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException; + +import java.io.Serializable; +import java.security.InvalidParameterException; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; + +/** Base class for all event types in the system. */ +public abstract class Event implements Serializable { Review Comment: Why implementing `Serializable`? Does jackson require it? ########## api/src/main/java/org/apache/flink/agents/api/utils/SerializableChecker.java: ########## @@ -0,0 +1,34 @@ +/* + * 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.api.utils; + +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; + +/** Util class for checking object serializable. */ +public final class SerializableChecker { Review Comment: If we do the check in runtime, we should also move this out of the api module. ########## api/src/test/java/org/apache/flink/agents/api/TestEvent.java: ########## @@ -0,0 +1,71 @@ +/* + * 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.api; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.security.InvalidParameterException; + +/** Test Event. */ +public class TestEvent { + public static class SerializableClass { + private final int a; + + public SerializableClass(int a) { + this.a = a; + } + + public int getA() { + return a; + } + } + + public static class UnserializableClass { + private final int a; + + public UnserializableClass(int a) { + this.a = a; + } + } + + @Test + public void testEventInitSerializable() { + InputEvent event = new InputEvent(new SerializableClass(1)); Review Comment: Unused local variable `event` ########## plan/src/main/java/org/apache/flink/agents/plan/JavaFunction.java: ########## @@ -0,0 +1,58 @@ +/* + * 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.plan; + +import java.lang.reflect.Method; +import java.util.Arrays; + +/** Represent a Java function. */ +public class JavaFunction implements Function { + private final String qualName; + private final String methodName; + private final Class<?>[] parameterTypes; + + public JavaFunction(String qualName, String methodName, Class<?>[] parameterTypes) { + this.qualName = qualName; + this.methodName = methodName; + this.parameterTypes = parameterTypes; + } + + @Override + public Object call(Object... args) throws Exception { + Class<?> clazz = Class.forName(qualName); + Method method = clazz.getMethod(methodName, parameterTypes); Review Comment: We should only do this once. ########## api/src/main/java/org/apache/flink/agents/api/OutputEvent.java: ########## @@ -0,0 +1,40 @@ +/* + * 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.api; + +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException; + +/** + * Event representing a result from workflow. By generating an OutputEvent, actions can emit output + * data. + */ +public class OutputEvent extends Event { + private static final long serialVersionUID = 1L; + private final Object output; + + public OutputEvent(Object output) throws JsonProcessingException { Review Comment: Unused exception. ########## plan/src/main/java/org/apache/flink/agents/plan/Action.java: ########## @@ -0,0 +1,55 @@ +/* + * 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.plan; + +import org.apache.flink.agents.api.Event; + +import java.io.Serializable; +import java.util.List; + +/** + * Representation of a workflow action with event listening and function execution. + * + * <p>This class encapsulates a named workflow action that listens for specific event types and + * executes an associated function when those events occur. + */ +public class Action implements Serializable { + private final String name; + private final Function exec; + private final List<Class<?>> listenEventTypes; Review Comment: ```suggestion private final List<Class<? extends Event>> listenEventTypes; ``` ########## api/src/test/java/org/apache/flink/agents/api/TestEvent.java: ########## @@ -0,0 +1,71 @@ +/* + * 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.api; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.security.InvalidParameterException; + +/** Test Event. */ +public class TestEvent { + public static class SerializableClass { + private final int a; + + public SerializableClass(int a) { + this.a = a; + } + + public int getA() { + return a; + } + } + + public static class UnserializableClass { Review Comment: How is this unserializable? ########## api/src/test/java/org/apache/flink/agents/api/TestEvent.java: ########## @@ -0,0 +1,71 @@ +/* + * 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.api; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.security.InvalidParameterException; + +/** Test Event. */ +public class TestEvent { + public static class SerializableClass { + private final int a; + + public SerializableClass(int a) { + this.a = a; + } + + public int getA() { + return a; + } Review Comment: unused ########## plan/src/main/java/org/apache/flink/agents/plan/WorkflowPlan.java: ########## @@ -0,0 +1,39 @@ +/* + * 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.plan; + +import java.util.List; +import java.util.Map; + +/** Workflow plan compiled from user defined workflow. */ +public class WorkflowPlan { + private final Map<Class<?>, List<Action>> actions; Review Comment: ```suggestion private final Map<Class<? extends Event>, List<Action>> actions; ``` -- 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]
