[WIP][ISSUE #4426] Optimize EventMesh Bridge with QUIC

.asf.yaml

@@ -47,6 +47,25 @@ github:
       required_pull_request_reviews:
         dismiss_stale_reviews: true
         required_approving_review_count: 2
+  # Attempt to make the auto-generated github emails more easily readable in email clients.
+  custom_subjects:
+    new_pr: "[PR] {title} ({repository})"
+    close_pr: "Re: [PR] {title} ({repository})"
+    comment_pr: "Re: [PR] {title} ({repository})"
+    diffcomment: "Re: [PR] {title} ({repository})"
+    merge_pr: "Re: [PR] {title} ({repository})"
+    new_issue: "[I] {title} ({repository})"
+    comment_issue: "Re: [I] {title} ({repository})"
+    close_issue: "Re: [I] {title} ({repository})"
+    catchall: "[GH] {title} ({repository})"
+    new_discussion: "[D] {title} ({repository})"
+    edit_discussion: "Re: [D] {title} ({repository})"
+    close_discussion: "Re: [D] {title} ({repository})"
+    close_discussion_with_comment: "Re: [D] {title} ({repository})"
+    reopen_discussion: "Re: [D] {title} ({repository})"
+    new_comment_discussion: "Re: [D] {title} ({repository})"
+    edit_comment_discussion: "Re: [D] {title} ({repository})"
+    delete_comment_discussion: "Re: [D] {title} ({repository})"
 notifications:
     commits: [email protected]
     # Send all issue emails (new, closed, comments) to issues@

.github/workflows/greetings.yml

@@ -43,7 +43,7 @@ jobs:
 
             |WeChat Assistant|WeChat Public Account|Slack|
             |-|-|-|
-            |<img src="https://github.com/apache/eventmesh/blob/master/resources/wechat-assistant.jpg?raw=true" width="128"/>|<img src="https://github.com/apache/eventmesh/blob/master/resources/wechat-official.jpg?raw=true" width="128"/>|[Join Slack Chat](https://join.slack.com/t/apacheeventmesh/shared_invite/zt-1t1816dli-I0t3OE~IpdYWrZbIWhMbXg)|
+            |<img src="https://github.com/apache/eventmesh/blob/master/resources/wechat-assistant.jpg?raw=true" width="128"/>|<img src="https://github.com/apache/eventmesh/blob/master/resources/wechat-official.jpg?raw=true" width="128"/>|[Join Slack Chat](https://join.slack.com/t/the-asf/shared_invite/zt-1y375qcox-UW1898e4kZE_pqrNsrBM2g)|
 
             Mailing Lists:
             | Name | Description |Subscribe	|Unsubscribe|Archive
@@ -65,7 +65,7 @@ jobs:
 
             |WeChat Assistant|WeChat Public Account|Slack|
             |-|-|-|
-            |<img src="https://github.com/apache/eventmesh/blob/master/resources/wechat-assistant.jpg?raw=true" width="128"/>|<img src="https://github.com/apache/eventmesh/blob/master/resources/wechat-official.jpg?raw=true" width="128"/>|[Join Slack Chat](https://join.slack.com/t/apacheeventmesh/shared_invite/zt-1t1816dli-I0t3OE~IpdYWrZbIWhMbXg)|
+            |<img src="https://github.com/apache/eventmesh/blob/master/resources/wechat-assistant.jpg?raw=true" width="128"/>|<img src="https://github.com/apache/eventmesh/blob/master/resources/wechat-official.jpg?raw=true" width="128"/>|[Join Slack Chat](https://join.slack.com/t/the-asf/shared_invite/zt-1y375qcox-UW1898e4kZE_pqrNsrBM2g)|
 
             Mailing Lists:
             | Name | Description |Subscribe	|Unsubscribe|Archive

.gitignore

@@ -11,6 +11,7 @@ logs
 *.log
 *.log.*
 .idea
+!/.idea/icon.png
 build
 .classpath
 .project

README.md

@@ -11,7 +11,7 @@
 
 [![License](https://img.shields.io/github/license/apache/eventmesh?style=for-the-badge)](https://www.apache.org/licenses/LICENSE-2.0.html)
 [![GitHub Release](https://img.shields.io/github/v/release/apache/eventmesh?style=for-the-badge)](https://github.com/apache/eventmesh/releases)
-[![Slack Status](https://img.shields.io/badge/slack-join_chat-blue.svg?logo=slack&style=for-the-badge)](https://join.slack.com/t/apacheeventmesh/shared_invite/zt-1vpgrvm5u-hh~5H9Aj4TGThaP9tusFfw)
+[![Slack Status](https://img.shields.io/badge/slack-join_chat-blue.svg?logo=slack&style=for-the-badge)](https://join.slack.com/t/the-asf/shared_invite/zt-1y375qcox-UW1898e4kZE_pqrNsrBM2g)
 
 
 [📦 Documentation](https://eventmesh.apache.org/docs/introduction) |
@@ -108,7 +108,7 @@ Apache EventMesh is licensed under the [Apache License, Version 2.0](http://www.
 
 |WeChat Assistant|WeChat Public Account|Slack|
 |-|-|-|
-|<img src="resources/wechat-assistant.jpg" width="128"/>|<img src="resources/wechat-official.jpg" width="128"/>|[Join Slack Chat](https://join.slack.com/t/apacheeventmesh/shared_invite/zt-1vpgrvm5u-hh~5H9Aj4TGThaP9tusFfw)|
+|<img src="resources/wechat-assistant.jpg" width="128"/>|<img src="resources/wechat-official.jpg" width="128"/>|[Join Slack Chat](https://join.slack.com/t/the-asf/shared_invite/zt-1y375qcox-UW1898e4kZE_pqrNsrBM2g)(Please open an issue if this link is expired)|
 
 Bi-weekly meeting : [#Tencent meeting](https://meeting.tencent.com/dm/wes6Erb9ioVV) : 346-6926-0133
 

build.gradle

@@ -21,6 +21,8 @@ import java.util.concurrent.TimeUnit
 
 buildscript {
     repositories {
+        mavenLocal()
+        mavenCentral()
         maven {
             url "https://maven.aliyun.com/repository/public"
         }
@@ -84,10 +86,15 @@ allprojects {
         configDirectory = file("${rootDir}/style")
     }
 
-    checkstyleMain.exclude '**/org/apache/eventmesh/client/grpc/protos**'
+    checkstyleMain.exclude('**/org/apache/eventmesh/client/grpc/protos**')
+                  .exclude('**/org/apache/eventmesh/common/protocol/grpc/cloudevents**')
+                  .exclude('**/org/apache/eventmesh/connector/openfunction/client/EventMeshGrpcService**')
+                  .exclude('**/org/apache/eventmesh/connector/openfunction/client/CallbackServiceGrpc**')
 
     dependencies {
         repositories {
+            mavenLocal()
+            mavenCentral()
             maven {
                 url "https://maven.aliyun.com/repository/public"
             }
@@ -361,9 +368,9 @@ subprojects {
     }
 
     repositories {
-        maven { url "https://maven.aliyun.com/repository/public" }
-        mavenCentral()
         mavenLocal()
+        mavenCentral()
+        maven { url "https://maven.aliyun.com/repository/public" }
     }
 
     configurations.all {
@@ -489,9 +496,11 @@ subprojects {
             dependency "org.powermock:powermock-module-junit4:2.0.2"
             dependency "org.powermock:powermock-api-mockito2:2.0.9"
 
-            dependency "io.cloudevents:cloudevents-core:2.2.0"
-            dependency "io.cloudevents:cloudevents-json-jackson:2.2.0"
+            dependency "io.cloudevents:cloudevents-core:2.4.2"
+            dependency "io.cloudevents:cloudevents-json-jackson:2.4.2"
+            dependency "io.cloudevents:cloudevents-protobuf:2.4.2"
 
+            dependency "io.grpc:grpc-core:${grpcVersion}"
             dependency "io.grpc:grpc-protobuf:${grpcVersion}"
             dependency "io.grpc:grpc-stub:${grpcVersion}"
             dependency "io.grpc:grpc-netty:${grpcVersion}"

docker/Dockerfile

@@ -29,7 +29,7 @@ WORKDIR /data/app/eventmesh
 COPY --from=builder /build/dist ./
 
 EXPOSE 10106 
-EXPOSE 10205 10105 10002
+EXPOSE 10205 10105 10000
 
 ENV DOCKER true
 ENV EVENTMESH_HOME /data/app/eventmesh

eventmesh-admin/eventmesh-admin-rocketmq/src/main/java/org/apache/eventmesh/admin/rocketmq/handler/TopicsHandler.java

@@ -64,7 +64,7 @@ public void createTopicHandler(HttpExchange httpExchange) throws IOException {
             String params = NetUtils.parsePostBody(httpExchange);
             TopicCreateRequest topicCreateRequest =
                 JsonUtils.parseObject(params, TopicCreateRequest.class);
-            String topic = topicCreateRequest.getName();
+            String topic = topicCreateRequest.getTopic();
 
             if (StringUtils.isBlank(topic)) {
                 result = "Create topic failed. Parameter topic not found.";

eventmesh-admin/eventmesh-admin-rocketmq/src/main/java/org/apache/eventmesh/admin/rocketmq/request/TopicCreateRequest.java

@@ -26,22 +26,19 @@
 @JsonIgnoreProperties(ignoreUnknown = true)
 public class TopicCreateRequest {
 
-    private String name;
+    private String topic;
 
     @JsonCreator
-    public TopicCreateRequest(@JsonProperty("name") String topic) {
-        super();
-        this.name = topic;
+    public TopicCreateRequest(@JsonProperty("topic") String topic) {
+        this.topic = topic;
     }
 
-    @JsonProperty("name")
-    public String getName() {
-        return this.name;
+    public String getTopic() {
+        return this.topic;
     }
 
-    @JsonProperty("name")
-    public void setName(String name) {
-        this.name = name;
+    public void setTopic(String topic) {
+        this.topic = topic;
     }
 
 }

eventmesh-admin/eventmesh-admin-rocketmq/src/test/java/org/apache/eventmesh/admin/rocketmq/request/TopicCreateRequestTest.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.eventmesh.admin.rocketmq.request;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Test;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+public class TopicCreateRequestTest {
+
+    @Test
+    public void testTopicCreateRequest() {
+        String name = "testTopic";
+        TopicCreateRequest topicCreateRequest = new TopicCreateRequest(name);
+
+        assertEquals(name, topicCreateRequest.getTopic());
+    }
+
+    @Test
+    public void testTopicCreateRequestSetName() {
+        TopicCreateRequest topicCreateRequest = new TopicCreateRequest(null);
+        assertNull(topicCreateRequest.getTopic());
+
+        String name = "testTopic";
+        topicCreateRequest.setTopic(name);
+        assertEquals(name, topicCreateRequest.getTopic());
+    }
+
+    @Test
+    public void testTopicCreateRequestSerialization() throws Exception {
+        String topic = "testTopic";
+        TopicCreateRequest topicCreateRequest = new TopicCreateRequest(topic);
+
+        ObjectMapper objectMapper = new ObjectMapper();
+        String json = objectMapper.writeValueAsString(topicCreateRequest);
+
+        assertTrue(json.contains("topic"));
+
+        TopicCreateRequest deserializedRequest = objectMapper.readValue(json, TopicCreateRequest.class);
+
+        assertEquals(topic, deserializedRequest.getTopic());
+    }
+
+}

eventmesh-common/build.gradle

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-def grpcVersion = '1.42.2'
+def grpcVersion = '1.43.2'
 
 dependencies {
     api "com.google.guava:guava"

eventmesh-connectors/sink-connector-rocketmq/src/main/java/org/apache/eventmesh/sink/connector/rocketmq/RocketMQSinkWorker.java → eventmesh-common/src/main/java/org/apache/eventmesh/common/enums/EventMeshDataContentType.java

@@ -15,16 +15,19 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.sink.connector.rocketmq;
+package org.apache.eventmesh.common.enums;
 
-import org.apache.eventmesh.openconnect.Application;
-import org.apache.eventmesh.sink.connector.rocketmq.connector.RocketMQSinkConnector;
+import lombok.Getter;
 
-public class RocketMQSinkWorker {
+public enum EventMeshDataContentType {
+    JSON("application/json");
 
-    public static void main(String[] args) throws Exception {
-
-        Application.run(RocketMQSinkConnector.class);
+    @Getter
+    private String code;
 
+    EventMeshDataContentType(String code) {
+        this.code = code;
     }
+
+
 }

eventmesh-connectors/source-connector-rocketmq/src/main/java/org/apache/eventmesh/source/connector/rocketmq/RocketMQSourceWorker.java → eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/HeartbeatItem.java

@@ -15,17 +15,21 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.source.connector.rocketmq;
+package org.apache.eventmesh.common.protocol;
 
-import org.apache.eventmesh.openconnect.Application;
-import org.apache.eventmesh.source.connector.rocketmq.connector.RocketMQSourceConnector;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
 
-public class RocketMQSourceWorker {
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@Builder
+public class HeartbeatItem {
 
-    public static void main(String[] args) throws Exception {
-
-        Application.run(RocketMQSourceConnector.class);
-
-    }
+    private String topic;
+
+    private String url;
 
 }

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/SubscriptionMode.java

@@ -26,7 +26,13 @@ public enum SubscriptionMode {
     /**
      * clustering
      */
-    CLUSTERING("CLUSTERING");
+    CLUSTERING("CLUSTERING"),
+
+    /**
+     * UNRECOGNIZED
+     */
+    UNRECOGNIZED("UNRECOGNIZED")
+    ;
 
     private final String mode;
 

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/SubscriptionType.java

@@ -25,7 +25,12 @@ public enum SubscriptionType {
     /**
      * ASYNC
      */
-    ASYNC("ASYNC");
+    ASYNC("ASYNC"),
+
+    /**
+     * UNRECOGNIZED
+     */
+    UNRECOGNIZED("UNRECOGNIZED");
 
     private final String type;
 

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/protos/ResponseOrBuilder.java → eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/cloudevents/CloudEventBatchOrBuilder.java

@@ -16,42 +16,35 @@
  */
 
 // Generated by the protocol buffer compiler.  DO NOT EDIT!
-// source: eventmesh-client.proto
+// source: eventmesh-cloudevents.proto
 
-package org.apache.eventmesh.common.protocol.grpc.protos;
+package org.apache.eventmesh.common.protocol.grpc.cloudevents;
 
-@SuppressWarnings({"all"})
-public interface ResponseOrBuilder extends
-    // @@protoc_insertion_point(interface_extends:eventmesh.common.protocol.grpc.Response)
+public interface CloudEventBatchOrBuilder extends
+    // @@protoc_insertion_point(interface_extends:org.apache.eventmesh.cloudevents.v1.CloudEventBatch)
     com.google.protobuf.MessageOrBuilder {
 
   /**
-   * <code>string respCode = 1;</code>
+   * <code>repeated .org.apache.eventmesh.cloudevents.v1.CloudEvent events = 1;</code>
    */
-  String getRespCode();
+  java.util.List<CloudEvent>
+      getEventsList();
   /**
-   * <code>string respCode = 1;</code>
+   * <code>repeated .org.apache.eventmesh.cloudevents.v1.CloudEvent events = 1;</code>
    */
-  com.google.protobuf.ByteString
-      getRespCodeBytes();
-
-  /**
-   * <code>string respMsg = 2;</code>
-   */
-  String getRespMsg();
+  CloudEvent getEvents(int index);
   /**
-   * <code>string respMsg = 2;</code>
+   * <code>repeated .org.apache.eventmesh.cloudevents.v1.CloudEvent events = 1;</code>
    */
-  com.google.protobuf.ByteString
-      getRespMsgBytes();
-
+  int getEventsCount();
   /**
-   * <code>string respTime = 3;</code>
+   * <code>repeated .org.apache.eventmesh.cloudevents.v1.CloudEvent events = 1;</code>
    */
-  String getRespTime();
+  java.util.List<? extends CloudEventOrBuilder>
+      getEventsOrBuilderList();
   /**
-   * <code>string respTime = 3;</code>
+   * <code>repeated .org.apache.eventmesh.cloudevents.v1.CloudEvent events = 1;</code>
    */
-  com.google.protobuf.ByteString
-      getRespTimeBytes();
+  CloudEventOrBuilder getEventsOrBuilder(
+      int index);
 }

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/cloudevents/CloudEventOrBuilder.java

@@ -0,0 +1,191 @@
+/*
+ * 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.
+ */
+
+// Generated by the protocol buffer compiler.  DO NOT EDIT!
+// source: eventmesh-cloudevents.proto
+
+package org.apache.eventmesh.common.protocol.grpc.cloudevents;
+
+public interface CloudEventOrBuilder extends
+    // @@protoc_insertion_point(interface_extends:org.apache.eventmesh.cloudevents.v1.CloudEvent)
+    com.google.protobuf.MessageOrBuilder {
+
+  /**
+   * <pre>
+   * Required Attributes
+   * </pre>
+   *
+   * <code>string id = 1;</code>
+   * @return The id.
+   */
+  String getId();
+  /**
+   * <pre>
+   * Required Attributes
+   * </pre>
+   *
+   * <code>string id = 1;</code>
+   * @return The bytes for id.
+   */
+  com.google.protobuf.ByteString
+      getIdBytes();
+
+  /**
+   * <pre>
+   * URI-reference
+   * </pre>
+   *
+   * <code>string source = 2;</code>
+   * @return The source.
+   */
+  String getSource();
+  /**
+   * <pre>
+   * URI-reference
+   * </pre>
+   *
+   * <code>string source = 2;</code>
+   * @return The bytes for source.
+   */
+  com.google.protobuf.ByteString
+      getSourceBytes();
+
+  /**
+   * <code>string spec_version = 3;</code>
+   * @return The specVersion.
+   */
+  String getSpecVersion();
+  /**
+   * <code>string spec_version = 3;</code>
+   * @return The bytes for specVersion.
+   */
+  com.google.protobuf.ByteString
+      getSpecVersionBytes();
+
+  /**
+   * <code>string type = 4;</code>
+   * @return The type.
+   */
+  String getType();
+  /**
+   * <code>string type = 4;</code>
+   * @return The bytes for type.
+   */
+  com.google.protobuf.ByteString
+      getTypeBytes();
+
+  /**
+   * <pre>
+   * Optional &amp; Extension Attributes
+   * </pre>
+   *
+   * <code>map&lt;string, .org.apache.eventmesh.cloudevents.v1.CloudEvent.CloudEventAttributeValue&gt; attributes = 5;</code>
+   */
+  int getAttributesCount();
+  /**
+   * <pre>
+   * Optional &amp; Extension Attributes
+   * </pre>
+   *
+   * <code>map&lt;string, .org.apache.eventmesh.cloudevents.v1.CloudEvent.CloudEventAttributeValue&gt; attributes = 5;</code>
+   */
+  boolean containsAttributes(
+      String key);
+  /**
+   * Use {@link #getAttributesMap()} instead.
+   */
+  @Deprecated
+  java.util.Map<String, CloudEvent.CloudEventAttributeValue>
+  getAttributes();
+  /**
+   * <pre>
+   * Optional &amp; Extension Attributes
+   * </pre>
+   *
+   * <code>map&lt;string, .org.apache.eventmesh.cloudevents.v1.CloudEvent.CloudEventAttributeValue&gt; attributes = 5;</code>
+   */
+  java.util.Map<String, CloudEvent.CloudEventAttributeValue>
+  getAttributesMap();
+  /**
+   * <pre>
+   * Optional &amp; Extension Attributes
+   * </pre>
+   *
+   * <code>map&lt;string, .org.apache.eventmesh.cloudevents.v1.CloudEvent.CloudEventAttributeValue&gt; attributes = 5;</code>
+   */
+
+  /* nullable */
+CloudEvent.CloudEventAttributeValue getAttributesOrDefault(
+      String key,
+      /* nullable */
+CloudEvent.CloudEventAttributeValue defaultValue);
+  /**
+   * <pre>
+   * Optional &amp; Extension Attributes
+   * </pre>
+   *
+   * <code>map&lt;string, .org.apache.eventmesh.cloudevents.v1.CloudEvent.CloudEventAttributeValue&gt; attributes = 5;</code>
+   */
+
+  CloudEvent.CloudEventAttributeValue getAttributesOrThrow(
+      String key);
+
+  /**
+   * <code>bytes binary_data = 6;</code>
+   * @return Whether the binaryData field is set.
+   */
+  boolean hasBinaryData();
+  /**
+   * <code>bytes binary_data = 6;</code>
+   * @return The binaryData.
+   */
+  com.google.protobuf.ByteString getBinaryData();
+
+  /**
+   * <code>string text_data = 7;</code>
+   * @return Whether the textData field is set.
+   */
+  boolean hasTextData();
+  /**
+   * <code>string text_data = 7;</code>
+   * @return The textData.
+   */
+  String getTextData();
+  /**
+   * <code>string text_data = 7;</code>
+   * @return The bytes for textData.
+   */
+  com.google.protobuf.ByteString
+      getTextDataBytes();
+
+  /**
+   * <code>.google.protobuf.Any proto_data = 8;</code>
+   * @return Whether the protoData field is set.
+   */
+  boolean hasProtoData();
+  /**
+   * <code>.google.protobuf.Any proto_data = 8;</code>
+   * @return The protoData.
+   */
+  com.google.protobuf.Any getProtoData();
+  /**
+   * <code>.google.protobuf.Any proto_data = 8;</code>
+   */
+  com.google.protobuf.AnyOrBuilder getProtoDataOrBuilder();
+
+  public CloudEvent.DataCase getDataCase();
+}

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/cloudevents/EventMeshCloudEvents.java

@@ -0,0 +1,122 @@
+/*
+ * 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.
+ */
+
+// Generated by the protocol buffer compiler.  DO NOT EDIT!
+// source: eventmesh-cloudevents.proto
+
+package org.apache.eventmesh.common.protocol.grpc.cloudevents;
+
+public final class EventMeshCloudEvents {
+  private EventMeshCloudEvents() {}
+  public static void registerAllExtensions(
+      com.google.protobuf.ExtensionRegistryLite registry) {
+  }
+
+  public static void registerAllExtensions(
+      com.google.protobuf.ExtensionRegistry registry) {
+    registerAllExtensions(
+        (com.google.protobuf.ExtensionRegistryLite) registry);
+  }
+  static final com.google.protobuf.Descriptors.Descriptor
+    internal_static_org_apache_eventmesh_cloudevents_v1_CloudEvent_descriptor;
+  static final
+    com.google.protobuf.GeneratedMessageV3.FieldAccessorTable
+      internal_static_org_apache_eventmesh_cloudevents_v1_CloudEvent_fieldAccessorTable;
+  static final com.google.protobuf.Descriptors.Descriptor
+    internal_static_org_apache_eventmesh_cloudevents_v1_CloudEvent_AttributesEntry_descriptor;
+  static final
+    com.google.protobuf.GeneratedMessageV3.FieldAccessorTable
+      internal_static_org_apache_eventmesh_cloudevents_v1_CloudEvent_AttributesEntry_fieldAccessorTable;
+  static final com.google.protobuf.Descriptors.Descriptor
+    internal_static_org_apache_eventmesh_cloudevents_v1_CloudEvent_CloudEventAttributeValue_descriptor;
+  static final
+    com.google.protobuf.GeneratedMessageV3.FieldAccessorTable
+      internal_static_org_apache_eventmesh_cloudevents_v1_CloudEvent_CloudEventAttributeValue_fieldAccessorTable;
+  static final com.google.protobuf.Descriptors.Descriptor
+    internal_static_org_apache_eventmesh_cloudevents_v1_CloudEventBatch_descriptor;
+  static final
+    com.google.protobuf.GeneratedMessageV3.FieldAccessorTable
+      internal_static_org_apache_eventmesh_cloudevents_v1_CloudEventBatch_fieldAccessorTable;
+
+  public static com.google.protobuf.Descriptors.FileDescriptor
+      getDescriptor() {
+    return descriptor;
+  }
+  private static  com.google.protobuf.Descriptors.FileDescriptor
+      descriptor;
+  static {
+    String[] descriptorData = {
+      "\n\033eventmesh-cloudevents.proto\022#org.apach" +
+      "e.eventmesh.cloudevents.v1\032\031google/proto" +
+      "buf/any.proto\032\037google/protobuf/timestamp" +
+      ".proto\"\324\004\n\nCloudEvent\022\n\n\002id\030\001 \001(\t\022\016\n\006sou" +
+      "rce\030\002 \001(\t\022\024\n\014spec_version\030\003 \001(\t\022\014\n\004type\030" +
+      "\004 \001(\t\022S\n\nattributes\030\005 \003(\0132?.org.apache.e" +
+      "ventmesh.cloudevents.v1.CloudEvent.Attri" +
+      "butesEntry\022\025\n\013binary_data\030\006 \001(\014H\000\022\023\n\ttex" +
+      "t_data\030\007 \001(\tH\000\022*\n\nproto_data\030\010 \001(\0132\024.goo" +
+      "gle.protobuf.AnyH\000\032{\n\017AttributesEntry\022\013\n" +
+      "\003key\030\001 \001(\t\022W\n\005value\030\002 \001(\0132H.org.apache.e" +
+      "ventmesh.cloudevents.v1.CloudEvent.Cloud" +
+      "EventAttributeValue:\0028\001\032\323\001\n\030CloudEventAt" +
+      "tributeValue\022\024\n\nce_boolean\030\001 \001(\010H\000\022\024\n\nce" +
+      "_integer\030\002 \001(\005H\000\022\023\n\tce_string\030\003 \001(\tH\000\022\022\n" +
+      "\010ce_bytes\030\004 \001(\014H\000\022\020\n\006ce_uri\030\005 \001(\tH\000\022\024\n\nc" +
+      "e_uri_ref\030\006 \001(\tH\000\0222\n\014ce_timestamp\030\007 \001(\0132" +
+      "\032.google.protobuf.TimestampH\000B\006\n\004attrB\006\n" +
+      "\004data\"R\n\017CloudEventBatch\022?\n\006events\030\001 \003(\013" +
+      "2/.org.apache.eventmesh.cloudevents.v1.C" +
+      "loudEventBO\n5org.apache.eventmesh.common" +
+      ".protocol.grpc.cloudeventsB\024EventMeshClo" +
+      "udEventsP\001b\006proto3"
+    };
+    descriptor = com.google.protobuf.Descriptors.FileDescriptor
+      .internalBuildGeneratedFileFrom(descriptorData,
+        new com.google.protobuf.Descriptors.FileDescriptor[] {
+          com.google.protobuf.AnyProto.getDescriptor(),
+          com.google.protobuf.TimestampProto.getDescriptor(),
+        });
+    internal_static_org_apache_eventmesh_cloudevents_v1_CloudEvent_descriptor =
+      getDescriptor().getMessageTypes().get(0);
+    internal_static_org_apache_eventmesh_cloudevents_v1_CloudEvent_fieldAccessorTable = new
+      com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
+        internal_static_org_apache_eventmesh_cloudevents_v1_CloudEvent_descriptor,
+        new String[] { "Id", "Source", "SpecVersion", "Type", "Attributes", "BinaryData", "TextData", "ProtoData", "Data", });
+    internal_static_org_apache_eventmesh_cloudevents_v1_CloudEvent_AttributesEntry_descriptor =
+      internal_static_org_apache_eventmesh_cloudevents_v1_CloudEvent_descriptor.getNestedTypes().get(0);
+    internal_static_org_apache_eventmesh_cloudevents_v1_CloudEvent_AttributesEntry_fieldAccessorTable = new
+      com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
+        internal_static_org_apache_eventmesh_cloudevents_v1_CloudEvent_AttributesEntry_descriptor,
+        new String[] { "Key", "Value", });
+    internal_static_org_apache_eventmesh_cloudevents_v1_CloudEvent_CloudEventAttributeValue_descriptor =
+      internal_static_org_apache_eventmesh_cloudevents_v1_CloudEvent_descriptor.getNestedTypes().get(1);
+    internal_static_org_apache_eventmesh_cloudevents_v1_CloudEvent_CloudEventAttributeValue_fieldAccessorTable = new
+      com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
+        internal_static_org_apache_eventmesh_cloudevents_v1_CloudEvent_CloudEventAttributeValue_descriptor,
+        new String[] { "CeBoolean", "CeInteger", "CeString", "CeBytes", "CeUri", "CeUriRef", "CeTimestamp", "Attr", });
+    internal_static_org_apache_eventmesh_cloudevents_v1_CloudEventBatch_descriptor =
+      getDescriptor().getMessageTypes().get(1);
+    internal_static_org_apache_eventmesh_cloudevents_v1_CloudEventBatch_fieldAccessorTable = new
+      com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
+        internal_static_org_apache_eventmesh_cloudevents_v1_CloudEventBatch_descriptor,
+        new String[] { "Events", });
+    com.google.protobuf.AnyProto.getDescriptor();
+    com.google.protobuf.TimestampProto.getDescriptor();
+  }
+
+  // @@protoc_insertion_point(outer_class_scope)
+}

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/cloudevents/EventMeshGrpcService.java

@@ -0,0 +1,87 @@
+/*
+ * 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.
+ */
+
+// Generated by the protocol buffer compiler.  DO NOT EDIT!
+// source: eventmesh-service.proto
+
+package org.apache.eventmesh.common.protocol.grpc.cloudevents;
+
+public final class EventMeshGrpcService {
+  private EventMeshGrpcService() {}
+  public static void registerAllExtensions(
+      com.google.protobuf.ExtensionRegistryLite registry) {
+  }
+
+  public static void registerAllExtensions(
+      com.google.protobuf.ExtensionRegistry registry) {
+    registerAllExtensions(
+        (com.google.protobuf.ExtensionRegistryLite) registry);
+  }
+
+  public static com.google.protobuf.Descriptors.FileDescriptor
+      getDescriptor() {
+    return descriptor;
+  }
+  private static  com.google.protobuf.Descriptors.FileDescriptor
+      descriptor;
+  static {
+    String[] descriptorData = {
+      "\n\027eventmesh-service.proto\022#org.apache.ev" +
+      "entmesh.cloudevents.v1\032\033google/protobuf/" +
+      "empty.proto\032\033eventmesh-cloudevents.proto" +
+      "2\246\004\n\020PublisherService\022k\n\007publish\022/.org.a" +
+      "pache.eventmesh.cloudevents.v1.CloudEven" +
+      "t\032/.org.apache.eventmesh.cloudevents.v1." +
+      "CloudEvent\022p\n\014publishReply\022/.org.apache." +
+      "eventmesh.cloudevents.v1.CloudEvent\032/.or" +
+      "g.apache.eventmesh.cloudevents.v1.CloudE" +
+      "vent\022X\n\rpublishOneWay\022/.org.apache.event" +
+      "mesh.cloudevents.v1.CloudEvent\032\026.google." +
+      "protobuf.Empty\022u\n\014batchPublish\0224.org.apa" +
+      "che.eventmesh.cloudevents.v1.CloudEventB" +
+      "atch\032/.org.apache.eventmesh.cloudevents." +
+      "v1.CloudEvent\022b\n\022batchPublishOneWay\0224.or" +
+      "g.apache.eventmesh.cloudevents.v1.CloudE" +
+      "ventBatch\032\026.google.protobuf.Empty2\352\002\n\017Co" +
+      "nsumerService\022m\n\tsubscribe\022/.org.apache." +
+      "eventmesh.cloudevents.v1.CloudEvent\032/.or" +
+      "g.apache.eventmesh.cloudevents.v1.CloudE" +
+      "vent\022w\n\017subscribeStream\022/.org.apache.eve" +
+      "ntmesh.cloudevents.v1.CloudEvent\032/.org.a" +
+      "pache.eventmesh.cloudevents.v1.CloudEven" +
+      "t(\0010\001\022o\n\013unsubscribe\022/.org.apache.eventm" +
+      "esh.cloudevents.v1.CloudEvent\032/.org.apac" +
+      "he.eventmesh.cloudevents.v1.CloudEvent2\201" +
+      "\001\n\020HeartbeatService\022m\n\theartbeat\022/.org.a" +
+      "pache.eventmesh.cloudevents.v1.CloudEven" +
+      "t\032/.org.apache.eventmesh.cloudevents.v1." +
+      "CloudEventBO\n5org.apache.eventmesh.commo" +
+      "n.protocol.grpc.cloudeventsB\024EventMeshGr" +
+      "pcServiceP\001b\006proto3"
+    };
+    descriptor = com.google.protobuf.Descriptors.FileDescriptor
+      .internalBuildGeneratedFileFrom(descriptorData,
+        new com.google.protobuf.Descriptors.FileDescriptor[] {
+          com.google.protobuf.EmptyProto.getDescriptor(),
+          EventMeshCloudEvents.getDescriptor(),
+        });
+    com.google.protobuf.EmptyProto.getDescriptor();
+    EventMeshCloudEvents.getDescriptor();
+  }
+
+  // @@protoc_insertion_point(outer_class_scope)
+}

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/protos/HeartbeatServiceGrpc.java → eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/cloudevents/HeartbeatServiceGrpc.java

@@ -15,81 +15,96 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.common.protocol.grpc.protos;
+package org.apache.eventmesh.common.protocol.grpc.cloudevents;
 
 import static io.grpc.MethodDescriptor.generateFullMethodName;
-import static io.grpc.stub.ClientCalls.asyncUnaryCall;
-import static io.grpc.stub.ClientCalls.blockingUnaryCall;
-import static io.grpc.stub.ClientCalls.futureUnaryCall;
-import static io.grpc.stub.ServerCalls.asyncUnaryCall;
-import static io.grpc.stub.ServerCalls.asyncUnimplementedUnaryCall;
 
 /**
  */
-@SuppressWarnings({"all"})
 @javax.annotation.Generated(
-    value = "by gRPC proto compiler (version 1.42.2)",
-    comments = "Source: eventmesh-client.proto")
+    value = "by gRPC proto compiler (version 1.43.2)",
+    comments = "Source: eventmesh-service.proto")
+@io.grpc.stub.annotations.GrpcGenerated
 public final class HeartbeatServiceGrpc {
 
   private HeartbeatServiceGrpc() {}
 
-  public static final String SERVICE_NAME = "eventmesh.common.protocol.grpc.HeartbeatService";
+  public static final String SERVICE_NAME = "org.apache.eventmesh.cloudevents.v1.HeartbeatService";
 
   // Static method descriptors that strictly reflect the proto.
-  private static volatile io.grpc.MethodDescriptor<Heartbeat,
-      Response> getHeartbeatMethod;
+  private static volatile io.grpc.MethodDescriptor<CloudEvent,
+      CloudEvent> getHeartbeatMethod;
 
   @io.grpc.stub.annotations.RpcMethod(
       fullMethodName = SERVICE_NAME + '/' + "heartbeat",
-      requestType = Heartbeat.class,
-      responseType = Response.class,
+      requestType = CloudEvent.class,
+      responseType = CloudEvent.class,
       methodType = io.grpc.MethodDescriptor.MethodType.UNARY)
-  public static io.grpc.MethodDescriptor<Heartbeat,
-      Response> getHeartbeatMethod() {
-    io.grpc.MethodDescriptor<Heartbeat, Response> getHeartbeatMethod;
+  public static io.grpc.MethodDescriptor<CloudEvent,
+      CloudEvent> getHeartbeatMethod() {
+    io.grpc.MethodDescriptor<CloudEvent, CloudEvent> getHeartbeatMethod;
     if ((getHeartbeatMethod = HeartbeatServiceGrpc.getHeartbeatMethod) == null) {
       synchronized (HeartbeatServiceGrpc.class) {
         if ((getHeartbeatMethod = HeartbeatServiceGrpc.getHeartbeatMethod) == null) {
-          HeartbeatServiceGrpc.getHeartbeatMethod = getHeartbeatMethod = 
-              io.grpc.MethodDescriptor.<Heartbeat, Response>newBuilder()
+          HeartbeatServiceGrpc.getHeartbeatMethod = getHeartbeatMethod =
+              io.grpc.MethodDescriptor.<CloudEvent, CloudEvent>newBuilder()
               .setType(io.grpc.MethodDescriptor.MethodType.UNARY)
-              .setFullMethodName(generateFullMethodName(
-                  "eventmesh.common.protocol.grpc.HeartbeatService", "heartbeat"))
+              .setFullMethodName(generateFullMethodName(SERVICE_NAME, "heartbeat"))
               .setSampledToLocalTracing(true)
               .setRequestMarshaller(io.grpc.protobuf.ProtoUtils.marshaller(
-                  Heartbeat.getDefaultInstance()))
+                  CloudEvent.getDefaultInstance()))
               .setResponseMarshaller(io.grpc.protobuf.ProtoUtils.marshaller(
-                  Response.getDefaultInstance()))
-                  .setSchemaDescriptor(new HeartbeatServiceMethodDescriptorSupplier("heartbeat"))
-                  .build();
-          }
+                  CloudEvent.getDefaultInstance()))
+              .setSchemaDescriptor(new HeartbeatServiceMethodDescriptorSupplier("heartbeat"))
+              .build();
         }
-     }
-     return getHeartbeatMethod;
+      }
+    }
+    return getHeartbeatMethod;
   }
 
   /**
    * Creates a new async stub that supports all call types for the service
    */
   public static HeartbeatServiceStub newStub(io.grpc.Channel channel) {
-    return new HeartbeatServiceStub(channel);
+    io.grpc.stub.AbstractStub.StubFactory<HeartbeatServiceStub> factory =
+      new io.grpc.stub.AbstractStub.StubFactory<HeartbeatServiceStub>() {
+        @Override
+        public HeartbeatServiceStub newStub(io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
+          return new HeartbeatServiceStub(channel, callOptions);
+        }
+      };
+    return HeartbeatServiceStub.newStub(factory, channel);
   }
 
   /**
    * Creates a new blocking-style stub that supports unary and streaming output calls on the service
    */
   public static HeartbeatServiceBlockingStub newBlockingStub(
       io.grpc.Channel channel) {
-    return new HeartbeatServiceBlockingStub(channel);
+    io.grpc.stub.AbstractStub.StubFactory<HeartbeatServiceBlockingStub> factory =
+      new io.grpc.stub.AbstractStub.StubFactory<HeartbeatServiceBlockingStub>() {
+        @Override
+        public HeartbeatServiceBlockingStub newStub(io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
+          return new HeartbeatServiceBlockingStub(channel, callOptions);
+        }
+      };
+    return HeartbeatServiceBlockingStub.newStub(factory, channel);
   }
 
   /**
    * Creates a new ListenableFuture-style stub that supports unary calls on the service
    */
   public static HeartbeatServiceFutureStub newFutureStub(
       io.grpc.Channel channel) {
-    return new HeartbeatServiceFutureStub(channel);
+    io.grpc.stub.AbstractStub.StubFactory<HeartbeatServiceFutureStub> factory =
+      new io.grpc.stub.AbstractStub.StubFactory<HeartbeatServiceFutureStub>() {
+        @Override
+        public HeartbeatServiceFutureStub newStub(io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
+          return new HeartbeatServiceFutureStub(channel, callOptions);
+        }
+      };
+    return HeartbeatServiceFutureStub.newStub(factory, channel);
   }
 
   /**
@@ -98,100 +113,88 @@ public static abstract class HeartbeatServiceImplBase implements io.grpc.Bindabl
 
     /**
      */
-    public void heartbeat(Heartbeat request,
-                          io.grpc.stub.StreamObserver<Response> responseObserver) {
-      asyncUnimplementedUnaryCall(getHeartbeatMethod(), responseObserver);
+    public void heartbeat(CloudEvent request,
+        io.grpc.stub.StreamObserver<CloudEvent> responseObserver) {
+      io.grpc.stub.ServerCalls.asyncUnimplementedUnaryCall(getHeartbeatMethod(), responseObserver);
     }
 
     @Override public final io.grpc.ServerServiceDefinition bindService() {
       return io.grpc.ServerServiceDefinition.builder(getServiceDescriptor())
           .addMethod(
             getHeartbeatMethod(),
-            asyncUnaryCall(
+            io.grpc.stub.ServerCalls.asyncUnaryCall(
               new MethodHandlers<
-                Heartbeat,
-                Response>(
+                CloudEvent,
+                CloudEvent>(
                   this, METHODID_HEARTBEAT)))
           .build();
     }
   }
 
   /**
    */
-  public static final class HeartbeatServiceStub extends io.grpc.stub.AbstractStub<HeartbeatServiceStub> {
-    private HeartbeatServiceStub(io.grpc.Channel channel) {
-      super(channel);
-    }
-
-    private HeartbeatServiceStub(io.grpc.Channel channel,
-        io.grpc.CallOptions callOptions) {
+  public static final class HeartbeatServiceStub extends io.grpc.stub.AbstractAsyncStub<HeartbeatServiceStub> {
+    private HeartbeatServiceStub(
+        io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
       super(channel, callOptions);
     }
 
     @Override
-    protected HeartbeatServiceStub build(io.grpc.Channel channel,
-        io.grpc.CallOptions callOptions) {
+    protected HeartbeatServiceStub build(
+        io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
       return new HeartbeatServiceStub(channel, callOptions);
     }
 
     /**
      */
-    public void heartbeat(Heartbeat request,
-                          io.grpc.stub.StreamObserver<Response> responseObserver) {
-      asyncUnaryCall(
+    public void heartbeat(CloudEvent request,
+        io.grpc.stub.StreamObserver<CloudEvent> responseObserver) {
+      io.grpc.stub.ClientCalls.asyncUnaryCall(
           getChannel().newCall(getHeartbeatMethod(), getCallOptions()), request, responseObserver);
     }
   }
 
   /**
    */
-  public static final class HeartbeatServiceBlockingStub extends io.grpc.stub.AbstractStub<HeartbeatServiceBlockingStub> {
-    private HeartbeatServiceBlockingStub(io.grpc.Channel channel) {
-      super(channel);
-    }
-
-    private HeartbeatServiceBlockingStub(io.grpc.Channel channel,
-        io.grpc.CallOptions callOptions) {
+  public static final class HeartbeatServiceBlockingStub extends io.grpc.stub.AbstractBlockingStub<HeartbeatServiceBlockingStub> {
+    private HeartbeatServiceBlockingStub(
+        io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
       super(channel, callOptions);
     }
 
     @Override
-    protected HeartbeatServiceBlockingStub build(io.grpc.Channel channel,
-        io.grpc.CallOptions callOptions) {
+    protected HeartbeatServiceBlockingStub build(
+        io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
       return new HeartbeatServiceBlockingStub(channel, callOptions);
     }
 
     /**
      */
-    public Response heartbeat(Heartbeat request) {
-      return blockingUnaryCall(
+    public CloudEvent heartbeat(CloudEvent request) {
+      return io.grpc.stub.ClientCalls.blockingUnaryCall(
           getChannel(), getHeartbeatMethod(), getCallOptions(), request);
     }
   }
 
   /**
    */
-  public static final class HeartbeatServiceFutureStub extends io.grpc.stub.AbstractStub<HeartbeatServiceFutureStub> {
-    private HeartbeatServiceFutureStub(io.grpc.Channel channel) {
-      super(channel);
-    }
-
-    private HeartbeatServiceFutureStub(io.grpc.Channel channel,
-        io.grpc.CallOptions callOptions) {
+  public static final class HeartbeatServiceFutureStub extends io.grpc.stub.AbstractFutureStub<HeartbeatServiceFutureStub> {
+    private HeartbeatServiceFutureStub(
+        io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
       super(channel, callOptions);
     }
 
     @Override
-    protected HeartbeatServiceFutureStub build(io.grpc.Channel channel,
-        io.grpc.CallOptions callOptions) {
+    protected HeartbeatServiceFutureStub build(
+        io.grpc.Channel channel, io.grpc.CallOptions callOptions) {
       return new HeartbeatServiceFutureStub(channel, callOptions);
     }
 
     /**
      */
-    public com.google.common.util.concurrent.ListenableFuture<Response> heartbeat(
-        Heartbeat request) {
-      return futureUnaryCall(
+    public com.google.common.util.concurrent.ListenableFuture<CloudEvent> heartbeat(
+        CloudEvent request) {
+      return io.grpc.stub.ClientCalls.futureUnaryCall(
           getChannel().newCall(getHeartbeatMethod(), getCallOptions()), request);
     }
   }
@@ -216,8 +219,8 @@ private static final class MethodHandlers<Req, Resp> implements
     public void invoke(Req request, io.grpc.stub.StreamObserver<Resp> responseObserver) {
       switch (methodId) {
         case METHODID_HEARTBEAT:
-          serviceImpl.heartbeat((Heartbeat) request,
-              (io.grpc.stub.StreamObserver<Response>) responseObserver);
+          serviceImpl.heartbeat((CloudEvent) request,
+              (io.grpc.stub.StreamObserver<CloudEvent>) responseObserver);
           break;
         default:
           throw new AssertionError();
@@ -241,7 +244,7 @@ private static abstract class HeartbeatServiceBaseDescriptorSupplier
 
     @Override
     public com.google.protobuf.Descriptors.FileDescriptor getFileDescriptor() {
-      return EventmeshGrpc.getDescriptor();
+      return EventMeshGrpcService.getDescriptor();
     }
 
     @Override

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/BatchMessageWrapper.java → eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/BatchEventMeshCloudEventWrapper.java

@@ -18,18 +18,18 @@
 package org.apache.eventmesh.common.protocol.grpc.common;
 
 import org.apache.eventmesh.common.protocol.ProtocolTransportObject;
-import org.apache.eventmesh.common.protocol.grpc.protos.BatchMessage;
+import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEventBatch;
 
-public class BatchMessageWrapper implements ProtocolTransportObject {
+public class BatchEventMeshCloudEventWrapper implements ProtocolTransportObject {
 
     private static final long serialVersionUID = -3296467364340663768L;
-    private final BatchMessage batchMessage;
+    private final CloudEventBatch cloudEventBatch;
 
-    public BatchMessageWrapper(BatchMessage batchMessage) {
-        this.batchMessage = batchMessage;
+    public BatchEventMeshCloudEventWrapper(CloudEventBatch cloudEventBatch) {
+        this.cloudEventBatch = cloudEventBatch;
     }
 
-    public BatchMessage getMessage() {
-        return batchMessage;
+    public CloudEventBatch getMessage() {
+        return cloudEventBatch;
     }
 }

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/ClientType.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.eventmesh.common.protocol.grpc.common;
+
+public enum ClientType {
+
+    PUB(1, "Client for publishing"),
+
+    SUB(2, "Client for subscribing");
+
+    private final int type;
+
+    private final String desc;
+
+    ClientType(int type, String desc) {
+        this.type = type;
+        this.desc = desc;
+    }
+
+    public static ClientType get(int type) {
+        for (ClientType clientType : ClientType.values()) {
+            if (clientType.type == type) {
+                return clientType;
+            }
+        }
+        return null;
+    }
+
+    public static boolean contains(Integer clientType) {
+        boolean flag = false;
+        for (ClientType ct : ClientType.values()) {
+            if (ct.type == clientType.intValue()) {
+                flag = true;
+                break;
+            }
+        }
+        return flag;
+    }
+
+    public int getType() {
+        return type;
+    }
+
+    public String getDesc() {
+        return desc;
+    }
+
+}

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/SimpleMessageWrapper.java → eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/EventMeshCloudEventWrapper.java

@@ -18,19 +18,19 @@
 package org.apache.eventmesh.common.protocol.grpc.common;
 
 import org.apache.eventmesh.common.protocol.ProtocolTransportObject;
-import org.apache.eventmesh.common.protocol.grpc.protos.SimpleMessage;
+import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent;
 
-public class SimpleMessageWrapper implements ProtocolTransportObject {
+public class EventMeshCloudEventWrapper implements ProtocolTransportObject {
 
     private static final long serialVersionUID = 4562321593749195797L;
 
-    private final SimpleMessage simpleMessage;
+    private final CloudEvent cloudEvent;
 
-    public SimpleMessageWrapper(SimpleMessage simpleMessage) {
-        this.simpleMessage = simpleMessage;
+    public EventMeshCloudEventWrapper(CloudEvent cloudEvent) {
+        this.cloudEvent = cloudEvent;
     }
 
-    public SimpleMessage getMessage() {
-        return simpleMessage;
+    public CloudEvent getMessage() {
+        return cloudEvent;
     }
 }

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/ProtoSupport.java

@@ -0,0 +1,46 @@
+/*
+ * 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.eventmesh.common.protocol.grpc.common;
+
+public class ProtoSupport {
+
+    private ProtoSupport() {
+
+    }
+
+    public static boolean isTextContent(String contentType) {
+
+        if (contentType == null) {
+            return false;
+        }
+
+        return contentType.startsWith("text/")
+            || "application/json".equals(contentType)
+            || "application/xml".equals(contentType)
+            || contentType.endsWith("+json")
+            || contentType.endsWith("+xml")
+            ;
+    }
+
+    public static boolean isProtoContent(String contentType) {
+        if (contentType == null) {
+            return false;
+        }
+        return "application/protobuf".equals(contentType);
+    }
+}

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/ProtocolKey.java

@@ -19,6 +19,7 @@
 
 public class ProtocolKey {
 
+    //EventMesh extensions
     public static final String ENV = "env";
     public static final String IDC = "idc";
     public static final String SYS = "sys";
@@ -27,16 +28,41 @@ public class ProtocolKey {
     public static final String USERNAME = "username";
     public static final String PASSWD = "passwd";
     public static final String LANGUAGE = "language";
-
     public static final String PROTOCOL_TYPE = "protocoltype";
     public static final String PROTOCOL_VERSION = "protocolversion";
     public static final String PROTOCOL_DESC = "protocoldesc";
-
     public static final String SEQ_NUM = "seqnum";
     public static final String UNIQUE_ID = "uniqueid";
     public static final String TTL = "ttl";
     public static final String PRODUCERGROUP = "producergroup";
+    public static final String CONSUMERGROUP = "consumergroup";
     public static final String TAG = "tag";
-
     public static final String CONTENT_TYPE = "contenttype";
+    public static final String PROPERTY_MESSAGE_CLUSTER = "cluster";
+    public static final String URL = "url";
+
+    public static final String CLIENT_TYPE = "clienttype";
+
+    public static final String GRPC_RESPONSE_CODE = "status_code";
+    public static final String GRPC_RESPONSE_MESSAGE = "response_message";
+    public static final String GRPC_RESPONSE_TIME = "time";
+
+    /**
+     * CloudEvents spec
+     *
+     * @see <a href="https://github.com/cloudevents/spec/blob/v1.0.2/cloudevents/spec.md#context-attributes">context-attributes</a>
+     */
+
+    //Required attributes
+    public static final String ID = "id";
+    public static final String SOURCE = "source";
+    public static final String SPECVERSION = "specversion";
+    public static final String TYPE = "type";
+
+    //Optional attributes
+    public static final String DATA_CONTENT_TYPE = "datacontenttype";
+    public static final String DATA_SCHEMA = "dataschema";
+    public static final String SUBJECT = "subject";
+    public static final String TIME = "time";
+    public static final String EVENT_DATA = "eventdata";
 }

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/Response.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.eventmesh.common.protocol.grpc.common;
+
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@EqualsAndHashCode
+@Builder
+@ToString
+public class Response {
+
+    private String respCode;
+
+    private String respMsg;
+
+    private String respTime;
+
+}

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/StatusCode.java

@@ -24,8 +24,8 @@ public enum StatusCode {
     EVENTMESH_REQUESTCODE_INVALID("2", "requestCode can't be null, or must be number, "),
     EVENTMESH_SEND_SYNC_MSG_ERR("3", "eventMesh send rr msg err, "),
     EVENTMESH_WAITING_RR_MSG_ERR("4", "eventMesh waiting rr msg err, "),
-    EVENTMESH_PROTOCOL_HEADER_ERR("6", "eventMesh protocol[header] err, "),
-    EVENTMESH_PROTOCOL_BODY_ERR("7", "eventMesh protocol[body] err, "),
+    EVENTMESH_PROTOCOL_HEADER_ERR("6", "eventMesh CloudEvent[header] err, "),
+    EVENTMESH_PROTOCOL_BODY_ERR("7", "eventMesh CloudEvent[body] err, "),
     EVENTMESH_STOP("8", "eventMesh will stop or had stopped, "),
     EVENTMESH_REJECT_BY_PROCESSOR_ERROR("9", "eventMesh reject by processor error, "),
     EVENTMESH_BATCH_PUBLISH_ERR("10", "eventMesh batch publish messages error, "),

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/common/SubscriptionReply.java

@@ -0,0 +1,57 @@
+/*
+ * 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.eventmesh.common.protocol.grpc.common;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class SubscriptionReply {
+
+    private String producerGroup;
+
+    private String topic;
+
+    private String content;
+
+    private String ttl;
+
+    private String uniqueId;
+
+    private String seqNum;
+
+    private String tag;
+
+    @Builder.Default
+    private Map<String, String> properties = new HashMap<>();
+
+    public void putAllProperties(Map<String, String> properties) {
+        if (null == properties || properties.isEmpty()) {
+            return;
+        }
+        properties.putAll(properties);
+    }
+}

eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/IPUtils.java

@@ -29,6 +29,7 @@
 import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Enumeration;
 import java.util.List;
 import java.util.regex.Matcher;
@@ -136,7 +137,7 @@ public static boolean isValidIPV4Address(String ip) {
         return m.matches();
     }
 
-    private static void getIpResult(List<String> ipv4Result, List<String> ipv6Result, Enumeration<InetAddress> en) {
+    private static void getIpResult(Collection<String> ipv4Result, Collection<String> ipv6Result, Enumeration<InetAddress> en) {
         while (en.hasMoreElements()) {
             final InetAddress address = en.nextElement();
             if (!address.isLoopbackAddress()) {

eventmesh-common/src/main/java/org/apache/eventmesh/common/utils/RandomStringUtils.java

@@ -19,6 +19,9 @@
 
 import org.apache.commons.text.RandomStringGenerator;
 
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+
 public class RandomStringUtils {
 
     private static final RandomStringGenerator RANDOM_NUM_GENERATOR = new RandomStringGenerator.Builder().withinRange('0', '9').build();
@@ -27,4 +30,8 @@ public static String generateNum(int length) {
         return RANDOM_NUM_GENERATOR.generate(length);
     }
 
+    public static String generateUUID() {
+        return new UUID(ThreadLocalRandom.current().nextLong(), ThreadLocalRandom.current().nextLong()).toString();
+    }
+
 }

eventmesh-common/src/test/java/org/apache/eventmesh/common/protocol/grpc/common/EventMeshCloudEventUtilsTest.java

@@ -0,0 +1,271 @@
+/*
+ * 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.eventmesh.common.protocol.grpc.common;
+
+import org.apache.eventmesh.common.enums.EventMeshProtocolType;
+import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent;
+import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent.CloudEventAttributeValue;
+
+import java.net.URI;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.google.protobuf.Timestamp;
+
+
+public class EventMeshCloudEventUtilsTest {
+
+    private CloudEvent cloudEvent;
+
+    @Before
+    public void init() {
+
+        final Map<String, CloudEventAttributeValue> attributeValueMap = new HashMap<>(64);
+        attributeValueMap.put(ProtocolKey.ENV, CloudEventAttributeValue.newBuilder().setCeString("dev").build());
+        attributeValueMap.put(ProtocolKey.IDC, CloudEventAttributeValue.newBuilder().setCeString("eventmesh").build());
+        attributeValueMap.put(ProtocolKey.IP, CloudEventAttributeValue.newBuilder().setCeString("127.0.0.1").build());
+        attributeValueMap.put(ProtocolKey.PID, CloudEventAttributeValue.newBuilder().setCeString("1243").build());
+        attributeValueMap.put(ProtocolKey.SYS, CloudEventAttributeValue.newBuilder().setCeString("eventmesh").build());
+        attributeValueMap.put(ProtocolKey.LANGUAGE, CloudEventAttributeValue.newBuilder().setCeString("java").build());
+        attributeValueMap.put(ProtocolKey.USERNAME, CloudEventAttributeValue.newBuilder().setCeString("mxsm").build());
+        attributeValueMap.put(ProtocolKey.PASSWD, CloudEventAttributeValue.newBuilder().setCeString("mxsm").build());
+        attributeValueMap.put(ProtocolKey.PROTOCOL_DESC, CloudEventAttributeValue.newBuilder().setCeString("version 1.0").build());
+        attributeValueMap.put(ProtocolKey.SEQ_NUM, CloudEventAttributeValue.newBuilder().setCeString("100").build());
+        attributeValueMap.put(ProtocolKey.UNIQUE_ID, CloudEventAttributeValue.newBuilder().setCeString("100").build());
+        attributeValueMap.put(ProtocolKey.TTL, CloudEventAttributeValue.newBuilder().setCeString("100").build());
+        attributeValueMap.put(ProtocolKey.PRODUCERGROUP, CloudEventAttributeValue.newBuilder().setCeString("mxsm_producer_group").build());
+        attributeValueMap.put(ProtocolKey.TAG, CloudEventAttributeValue.newBuilder().setCeString("tag").build());
+        attributeValueMap.put(ProtocolKey.SUBJECT, CloudEventAttributeValue.newBuilder().setCeString("topic").build());
+        attributeValueMap.put(ProtocolKey.CLIENT_TYPE, CloudEventAttributeValue.newBuilder().setCeInteger(ClientType.SUB.getType()).build());
+        attributeValueMap.put(ProtocolKey.GRPC_RESPONSE_CODE, CloudEventAttributeValue.newBuilder().setCeString("0").build());
+        Instant instant = LocalDateTime.of(2023, 4, 11, 19, 7, 0).toInstant(ZoneOffset.UTC);
+        attributeValueMap.put(ProtocolKey.GRPC_RESPONSE_TIME, CloudEventAttributeValue.newBuilder()
+            .setCeTimestamp(Timestamp.newBuilder().setSeconds(instant.getEpochSecond()).setNanos(instant.getNano()).build()).build());
+        attributeValueMap.put(ProtocolKey.GRPC_RESPONSE_MESSAGE, CloudEventAttributeValue.newBuilder().setCeString("0").build());
+        attributeValueMap.put(ProtocolKey.PROPERTY_MESSAGE_CLUSTER, CloudEventAttributeValue.newBuilder().setCeString("DefaultCluster").build());
+        attributeValueMap.put(ProtocolKey.URL, CloudEventAttributeValue.newBuilder().setCeString("http://127.0.0.1").build());
+        attributeValueMap.put(ProtocolKey.CONSUMERGROUP, CloudEventAttributeValue.newBuilder().setCeString("ConsumerGroup").build());
+        attributeValueMap.put(ProtocolKey.DATA_CONTENT_TYPE, CloudEventAttributeValue.newBuilder().setCeString("text/plain").build());
+        attributeValueMap.put(ProtocolKey.CONTENT_TYPE, CloudEventAttributeValue.newBuilder().setCeString("text/plain").build());
+        attributeValueMap.put(ProtocolKey.PROTOCOL_TYPE,
+            CloudEventAttributeValue.newBuilder().setCeString(EventMeshProtocolType.CLOUD_EVENTS.protocolTypeName()).build());
+        attributeValueMap.put(ProtocolKey.PROTOCOL_VERSION, CloudEventAttributeValue.newBuilder().setCeString("1.0").build());
+        cloudEvent = CloudEvent.newBuilder().putAllAttributes(attributeValueMap).setId("123").setSource(URI.create("/").toString())
+            .setType("eventmesh")
+            .setSpecVersion("1.0").setTextData("mxsm").build();
+    }
+
+    @Test
+    public void testGetEnv() {
+        Assert.assertEquals("dev", EventMeshCloudEventUtils.getEnv(cloudEvent));
+        Assert.assertEquals("dev", EventMeshCloudEventUtils.getEnv(cloudEvent, "test"));
+        Assert.assertEquals("test", EventMeshCloudEventUtils.getEnv(CloudEvent.newBuilder().build(), "test"));
+    }
+
+    @Test
+    public void testGetIdc() {
+        Assert.assertEquals("eventmesh", EventMeshCloudEventUtils.getIdc(cloudEvent));
+        Assert.assertEquals("eventmesh", EventMeshCloudEventUtils.getIdc(cloudEvent, "test"));
+        Assert.assertEquals("test", EventMeshCloudEventUtils.getIdc(CloudEvent.newBuilder().build(), "test"));
+    }
+
+    @Test
+    public void testGetSys() {
+        Assert.assertEquals("eventmesh", EventMeshCloudEventUtils.getSys(cloudEvent));
+        Assert.assertEquals("eventmesh", EventMeshCloudEventUtils.getSys(cloudEvent, "test"));
+        Assert.assertEquals("Linux", EventMeshCloudEventUtils.getSys(CloudEvent.newBuilder().build(), "Linux"));
+    }
+
+    @Test
+    public void testGetPid() {
+        Assert.assertEquals("1243", EventMeshCloudEventUtils.getPid(cloudEvent));
+        Assert.assertEquals("1243", EventMeshCloudEventUtils.getPid(cloudEvent, "test"));
+        Assert.assertEquals("987", EventMeshCloudEventUtils.getPid(CloudEvent.newBuilder().build(), "987"));
+    }
+
+    @Test
+    public void testGetIp() {
+        Assert.assertEquals("127.0.0.1", EventMeshCloudEventUtils.getIp(cloudEvent));
+        Assert.assertEquals("127.0.0.1", EventMeshCloudEventUtils.getIp(cloudEvent, "127.0.0.2"));
+        Assert.assertEquals("192.168.1.1", EventMeshCloudEventUtils.getIp(CloudEvent.newBuilder().build(), "192.168.1.1"));
+    }
+
+    @Test
+    public void testGetUserName() {
+        Assert.assertEquals("mxsm", EventMeshCloudEventUtils.getUserName(cloudEvent));
+        Assert.assertEquals("mxsm", EventMeshCloudEventUtils.getUserName(cloudEvent, "mxsm1"));
+        Assert.assertEquals("root", EventMeshCloudEventUtils.getUserName(CloudEvent.newBuilder().build(), "root"));
+    }
+
+    @Test
+    public void testGetPassword() {
+        Assert.assertEquals("mxsm", EventMeshCloudEventUtils.getPassword(cloudEvent));
+        Assert.assertEquals("mxsm", EventMeshCloudEventUtils.getPassword(cloudEvent, "mxsm1"));
+        Assert.assertEquals("root", EventMeshCloudEventUtils.getPassword(CloudEvent.newBuilder().build(), "root"));
+    }
+
+    @Test
+    public void testGetLanguage() {
+        Assert.assertEquals("java", EventMeshCloudEventUtils.getLanguage(cloudEvent));
+        Assert.assertEquals("java", EventMeshCloudEventUtils.getLanguage(cloudEvent, "Go"));
+        Assert.assertEquals("Go", EventMeshCloudEventUtils.getLanguage(CloudEvent.newBuilder().build(), "Go"));
+    }
+
+    @Test
+    public void testGetProtocolType() {
+        Assert.assertEquals(EventMeshProtocolType.CLOUD_EVENTS.protocolTypeName(), EventMeshCloudEventUtils.getProtocolType(cloudEvent));
+        Assert.assertEquals(EventMeshProtocolType.CLOUD_EVENTS.protocolTypeName(), EventMeshCloudEventUtils.getProtocolType(cloudEvent, "Go"));
+        Assert.assertEquals("eventmeshMessage", EventMeshCloudEventUtils.getProtocolType(CloudEvent.newBuilder().build(), "eventmeshMessage"));
+    }
+
+    @Test
+    public void testGetProtocolVersion() {
+        Assert.assertEquals("1.0", EventMeshCloudEventUtils.getProtocolVersion(cloudEvent));
+        Assert.assertEquals("1.0", EventMeshCloudEventUtils.getProtocolVersion(cloudEvent, "1.1"));
+        Assert.assertEquals("1.2", EventMeshCloudEventUtils.getProtocolVersion(CloudEvent.newBuilder().build(), "1.2"));
+    }
+
+    @Test
+    public void testGetProtocolDesc() {
+        Assert.assertEquals("version 1.0", EventMeshCloudEventUtils.getProtocolDesc(cloudEvent));
+        Assert.assertEquals("version 1.0", EventMeshCloudEventUtils.getProtocolDesc(cloudEvent, "version 1.1"));
+        Assert.assertEquals("version 1.2", EventMeshCloudEventUtils.getProtocolDesc(CloudEvent.newBuilder().build(), "version 1.2"));
+    }
+
+    @Test
+    public void testGetSeqNum() {
+        Assert.assertEquals("100", EventMeshCloudEventUtils.getSeqNum(cloudEvent));
+        Assert.assertEquals("100", EventMeshCloudEventUtils.getSeqNum(cloudEvent, "200"));
+        Assert.assertEquals("200", EventMeshCloudEventUtils.getSeqNum(CloudEvent.newBuilder().build(), "200"));
+    }
+
+    @Test
+    public void testGetUniqueId() {
+        Assert.assertEquals("100", EventMeshCloudEventUtils.getUniqueId(cloudEvent));
+        Assert.assertEquals("100", EventMeshCloudEventUtils.getUniqueId(cloudEvent, "200"));
+        Assert.assertEquals("200", EventMeshCloudEventUtils.getUniqueId(CloudEvent.newBuilder().build(), "200"));
+    }
+
+    @Test
+    public void testGetTtl() {
+        Assert.assertEquals("100", EventMeshCloudEventUtils.getTtl(cloudEvent));
+        Assert.assertEquals("100", EventMeshCloudEventUtils.getTtl(cloudEvent, "200"));
+        Assert.assertEquals("200", EventMeshCloudEventUtils.getTtl(CloudEvent.newBuilder().build(), "200"));
+    }
+
+    @Test
+    public void testGetProducerGroup() {
+        Assert.assertEquals("mxsm_producer_group", EventMeshCloudEventUtils.getProducerGroup(cloudEvent));
+        Assert.assertEquals("mxsm_producer_group", EventMeshCloudEventUtils.getProducerGroup(cloudEvent, "mxsm_producer_group"));
+        Assert.assertEquals("mxsm_producer_group1",
+            EventMeshCloudEventUtils.getProducerGroup(CloudEvent.newBuilder().build(), "mxsm_producer_group1"));
+    }
+
+    @Test
+    public void testGetTag() {
+        Assert.assertEquals("tag", EventMeshCloudEventUtils.getTag(cloudEvent));
+        Assert.assertEquals("tag", EventMeshCloudEventUtils.getTag(cloudEvent, "tag1"));
+        Assert.assertEquals("tag1", EventMeshCloudEventUtils.getTag(CloudEvent.newBuilder().build(), "tag1"));
+    }
+
+    @Test
+    public void testGetContentType() {
+        Assert.assertEquals("text/plain", EventMeshCloudEventUtils.getContentType(cloudEvent));
+        Assert.assertEquals("text/plain", EventMeshCloudEventUtils.getContentType(cloudEvent, "application/json"));
+        Assert.assertEquals("application/json", EventMeshCloudEventUtils.getContentType(CloudEvent.newBuilder().build(), "application/json"));
+    }
+
+    @Test
+    public void testGetSubject() {
+        Assert.assertEquals("topic", EventMeshCloudEventUtils.getSubject(cloudEvent));
+        Assert.assertEquals("topic", EventMeshCloudEventUtils.getSubject(cloudEvent, "topic12"));
+        Assert.assertEquals("mxsm-topic", EventMeshCloudEventUtils.getSubject(CloudEvent.newBuilder().build(), "mxsm-topic"));
+    }
+
+    @Test
+    public void testGetDataContentType() {
+        Assert.assertEquals("text/plain", EventMeshCloudEventUtils.getDataContentType(cloudEvent));
+        Assert.assertEquals("text/plain", EventMeshCloudEventUtils.getDataContentType(cloudEvent, "application/json"));
+        Assert.assertEquals("application/json", EventMeshCloudEventUtils.getDataContentType(CloudEvent.newBuilder().build(), "application/json"));
+    }
+
+    @Test
+    public void testGetResponseCode() {
+        Assert.assertEquals("0", EventMeshCloudEventUtils.getResponseCode(cloudEvent));
+        Assert.assertEquals("0", EventMeshCloudEventUtils.getResponseCode(cloudEvent, "1"));
+        Assert.assertEquals("1", EventMeshCloudEventUtils.getResponseCode(CloudEvent.newBuilder().build(), "1"));
+    }
+
+    @Test
+    public void testGetResponseMessage() {
+        Assert.assertEquals("0", EventMeshCloudEventUtils.getResponseMessage(cloudEvent));
+        Assert.assertEquals("0", EventMeshCloudEventUtils.getResponseMessage(cloudEvent, "1"));
+        Assert.assertEquals("1", EventMeshCloudEventUtils.getResponseMessage(CloudEvent.newBuilder().build(), "1"));
+    }
+
+    @Test
+    public void testGetResponseTime() {
+        Assert.assertEquals("2023-04-11T19:07Z", EventMeshCloudEventUtils.getResponseTime(cloudEvent));
+        Assert.assertEquals("2023-04-11T19:07Z", EventMeshCloudEventUtils.getResponseTime(cloudEvent, "2023-04-11 17:45:10"));
+        Assert.assertEquals("1970-01-01T00:00Z", EventMeshCloudEventUtils.getResponseTime(CloudEvent.newBuilder().build(), "1970-01-01T00:00Z"));
+    }
+
+    @Test
+    public void testGetCluster() {
+        Assert.assertEquals("DefaultCluster", EventMeshCloudEventUtils.getCluster(cloudEvent));
+        Assert.assertEquals("DefaultCluster", EventMeshCloudEventUtils.getCluster(cloudEvent, "DefaultCluster1"));
+        Assert.assertEquals("DefaultCluster1", EventMeshCloudEventUtils.getCluster(CloudEvent.newBuilder().build(), "DefaultCluster1"));
+    }
+
+    @Test
+    public void testGetConsumerGroup() {
+        Assert.assertEquals("ConsumerGroup", EventMeshCloudEventUtils.getConsumerGroup(cloudEvent));
+        Assert.assertEquals("ConsumerGroup", EventMeshCloudEventUtils.getConsumerGroup(cloudEvent, "ConsumerGroup111"));
+        Assert.assertEquals("ConsumerGroup111", EventMeshCloudEventUtils.getConsumerGroup(CloudEvent.newBuilder().build(), "ConsumerGroup111"));
+    }
+
+    @Test
+    public void testGetClientType() {
+        Assert.assertEquals(ClientType.SUB, EventMeshCloudEventUtils.getClientType(cloudEvent));
+        Assert.assertEquals(ClientType.SUB, EventMeshCloudEventUtils.getClientType(cloudEvent, ClientType.PUB));
+        Assert.assertEquals(ClientType.PUB, EventMeshCloudEventUtils.getClientType(CloudEvent.newBuilder().build(), ClientType.PUB));
+    }
+
+    @Test
+    public void testGetURL() {
+        Assert.assertEquals("http://127.0.0.1", EventMeshCloudEventUtils.getURL(cloudEvent));
+        Assert.assertEquals("http://127.0.0.1", EventMeshCloudEventUtils.getURL(cloudEvent, "http://127.0.0.2"));
+        Assert.assertEquals("http://127.0.0.2", EventMeshCloudEventUtils.getURL(CloudEvent.newBuilder().build(), "http://127.0.0.2"));
+    }
+
+    @Test
+    public void testGetDataContent() {
+        Assert.assertEquals("mxsm", EventMeshCloudEventUtils.getDataContent(cloudEvent));
+        Assert.assertEquals("mxsm", EventMeshCloudEventUtils.getDataContent(cloudEvent, "http://127.0.0.2"));
+        Assert.assertEquals("http://127.0.0.2", EventMeshCloudEventUtils.getDataContent(CloudEvent.newBuilder().build(), "http://127.0.0.2"));
+    }
+
+
+}

eventmesh-connectors/sink-connector-rocketmq/build.gradle → eventmesh-connectors/eventmesh-connector-openfunction/build.gradle

@@ -15,26 +15,23 @@
  * limitations under the License.
  */
 
-List rocketmq = [
-        "org.apache.rocketmq:rocketmq-client:$rocketmq_version",
-        "org.apache.rocketmq:rocketmq-broker:$rocketmq_version",
-        "org.apache.rocketmq:rocketmq-common:$rocketmq_version",
-        "org.apache.rocketmq:rocketmq-store:$rocketmq_version",
-        "org.apache.rocketmq:rocketmq-namesrv:$rocketmq_version",
-        "org.apache.rocketmq:rocketmq-tools:$rocketmq_version",
-        "org.apache.rocketmq:rocketmq-remoting:$rocketmq_version",
-        "org.apache.rocketmq:rocketmq-logging:$rocketmq_version",
-        "org.apache.rocketmq:rocketmq-test:$rocketmq_version",
-        "org.apache.rocketmq:rocketmq-srvutil:$rocketmq_version",
-        "org.apache.rocketmq:rocketmq-filter:$rocketmq_version",
-        "org.apache.rocketmq:rocketmq-acl:$rocketmq_version",
-        "org.apache.rocketmq:rocketmq-srvutil:$rocketmq_version",
-
-]
-
 dependencies {
+    implementation project(":eventmesh-common")
+    implementation project(":eventmesh-sdks:eventmesh-sdk-java")
     implementation project(":eventmesh-openconnect:eventmesh-openconnect-java")
-    implementation rocketmq
+
+    implementation "io.grpc:grpc-core"
+    implementation "io.grpc:grpc-protobuf"
+    implementation "io.grpc:grpc-stub"
+    implementation "io.grpc:grpc-netty"
+    implementation "io.grpc:grpc-netty-shaded"
+
+    api "com.fasterxml.jackson.core:jackson-databind"
+    api "com.fasterxml.jackson.core:jackson-core"
+    api "com.fasterxml.jackson.core:jackson-annotations"
+
+    implementation "javax.annotation:javax.annotation-api:1.3.2"
+
     compileOnly 'org.projectlombok:lombok'
     annotationProcessor 'org.projectlombok:lombok'
-}
+}

eventmesh-connectors/source-connector-rocketmq/gradle.properties → eventmesh-connectors/eventmesh-connector-openfunction/gradle.properties

@@ -13,5 +13,4 @@
 # 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.
-#
-rocketmq_version=4.9.5
+#

eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/client/CallbackService.java

@@ -0,0 +1,59 @@
+/*
+ * 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.eventmesh.connector.openfunction.client;
+
+import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent;
+import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent.CloudEventAttributeValue;
+import org.apache.eventmesh.common.protocol.grpc.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.grpc.common.StatusCode;
+
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+
+import io.grpc.stub.StreamObserver;
+
+import com.google.protobuf.Timestamp;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class CallbackService extends CallbackServiceGrpc.CallbackServiceImplBase {
+
+    @Override
+    public void onTopicEvent(CloudEvent cloudEvent, StreamObserver<CloudEvent> responseObserver) {
+        log.info("cloudevents: {}|data: {}", cloudEvent, cloudEvent.getTextData());
+
+        Instant instant = now();
+        CloudEvent.Builder builder = CloudEvent.newBuilder();
+        builder.putAttributes(ProtocolKey.GRPC_RESPONSE_CODE,
+                CloudEventAttributeValue.newBuilder().setCeString(StatusCode.SUCCESS.getRetCode()).build())
+            .putAttributes(ProtocolKey.GRPC_RESPONSE_MESSAGE,
+                CloudEventAttributeValue.newBuilder().setCeString(StatusCode.SUCCESS.getErrMsg()).build())
+            .putAttributes(ProtocolKey.GRPC_RESPONSE_TIME, CloudEventAttributeValue.newBuilder()
+                .setCeTimestamp(Timestamp.newBuilder().setSeconds(instant.getEpochSecond()).setNanos(instant.getNano()).build()).build());
+        responseObserver.onNext(builder.build());
+        responseObserver.onCompleted();
+    }
+
+    private static Instant now() {
+        return OffsetDateTime.of(LocalDateTime.now(ZoneId.systemDefault()), ZoneOffset.UTC).toInstant();
+    }
+}

eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/client/CallbackServiceGrpc.java

@@ -0,0 +1,268 @@
+/*
+ * 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.eventmesh.connector.openfunction.client;
+
+import static io.grpc.MethodDescriptor.generateFullMethodName;
+import static io.grpc.stub.ClientCalls.asyncUnaryCall;
+import static io.grpc.stub.ClientCalls.blockingUnaryCall;
+import static io.grpc.stub.ClientCalls.futureUnaryCall;
+import static io.grpc.stub.ServerCalls.asyncUnaryCall;
+import static io.grpc.stub.ServerCalls.asyncUnimplementedUnaryCall;
+
+/**
+ *
+ */
+@javax.annotation.Generated(
+    value = "by gRPC proto compiler (version 1.4.0)",
+    comments = "Source: callback-service.proto")
+@io.grpc.stub.annotations.GrpcGenerated
+public final class CallbackServiceGrpc {
+
+    private CallbackServiceGrpc() {
+    }
+
+    public static final String SERVICE_NAME = "org.apache.eventmesh.cloudevents.v1.CallbackService";
+
+    // Static method descriptors that strictly reflect the proto.
+    @io.grpc.ExperimentalApi("https://github.com/grpc/grpc-java/issues/1901")
+    public static final io.grpc.MethodDescriptor<org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent,
+        org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent> METHOD_ON_TOPIC_EVENT =
+        io.grpc.MethodDescriptor.<org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent, org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent>newBuilder()
+            .setType(io.grpc.MethodDescriptor.MethodType.UNARY)
+            .setFullMethodName(generateFullMethodName(
+                "org.apache.eventmesh.cloudevents.v1.CallbackService", "OnTopicEvent"))
+            .setRequestMarshaller(io.grpc.protobuf.ProtoUtils.marshaller(
+                org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent.getDefaultInstance()))
+            .setResponseMarshaller(io.grpc.protobuf.ProtoUtils.marshaller(
+                org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent.getDefaultInstance()))
+            .build();
+
+    /**
+     * Creates a new async stub that supports all call types for the service
+     */
+    public static CallbackServiceStub newStub(io.grpc.Channel channel) {
+        return new CallbackServiceStub(channel);
+    }
+
+    /**
+     * Creates a new blocking-style stub that supports unary and streaming output calls on the service
+     */
+    public static CallbackServiceBlockingStub newBlockingStub(
+        io.grpc.Channel channel) {
+        return new CallbackServiceBlockingStub(channel);
+    }
+
+    /**
+     * Creates a new ListenableFuture-style stub that supports unary calls on the service
+     */
+    public static CallbackServiceFutureStub newFutureStub(
+        io.grpc.Channel channel) {
+        return new CallbackServiceFutureStub(channel);
+    }
+
+    /**
+     *
+     */
+    public static abstract class CallbackServiceImplBase implements io.grpc.BindableService {
+
+        /**
+         * <pre>
+         * Subscribes events
+         * </pre>
+         */
+        public void onTopicEvent(org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent request,
+            io.grpc.stub.StreamObserver<org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent> responseObserver) {
+            asyncUnimplementedUnaryCall(METHOD_ON_TOPIC_EVENT, responseObserver);
+        }
+
+        @Override
+        public final io.grpc.ServerServiceDefinition bindService() {
+            return io.grpc.ServerServiceDefinition.builder(getServiceDescriptor())
+                .addMethod(
+                    METHOD_ON_TOPIC_EVENT,
+                    asyncUnaryCall(
+                        new MethodHandlers<
+                            org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent,
+                            org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent>(
+                            this, METHODID_ON_TOPIC_EVENT)))
+                .build();
+        }
+    }
+
+    /**
+     *
+     */
+    public static final class CallbackServiceStub extends io.grpc.stub.AbstractStub<CallbackServiceStub> {
+
+        private CallbackServiceStub(io.grpc.Channel channel) {
+            super(channel);
+        }
+
+        private CallbackServiceStub(io.grpc.Channel channel,
+            io.grpc.CallOptions callOptions) {
+            super(channel, callOptions);
+        }
+
+        @Override
+        protected CallbackServiceStub build(io.grpc.Channel channel,
+            io.grpc.CallOptions callOptions) {
+            return new CallbackServiceStub(channel, callOptions);
+        }
+
+        /**
+         * <pre>
+         * Subscribes events
+         * </pre>
+         */
+        public void onTopicEvent(org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent request,
+            io.grpc.stub.StreamObserver<org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent> responseObserver) {
+            asyncUnaryCall(
+                getChannel().newCall(METHOD_ON_TOPIC_EVENT, getCallOptions()), request, responseObserver);
+        }
+    }
+
+    /**
+     *
+     */
+    public static final class CallbackServiceBlockingStub extends io.grpc.stub.AbstractStub<CallbackServiceBlockingStub> {
+
+        private CallbackServiceBlockingStub(io.grpc.Channel channel) {
+            super(channel);
+        }
+
+        private CallbackServiceBlockingStub(io.grpc.Channel channel,
+            io.grpc.CallOptions callOptions) {
+            super(channel, callOptions);
+        }
+
+        @Override
+        protected CallbackServiceBlockingStub build(io.grpc.Channel channel,
+            io.grpc.CallOptions callOptions) {
+            return new CallbackServiceBlockingStub(channel, callOptions);
+        }
+
+        /**
+         * <pre>
+         * Subscribes events
+         * </pre>
+         */
+        public org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent onTopicEvent(
+            org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent request) {
+            return blockingUnaryCall(
+                getChannel(), METHOD_ON_TOPIC_EVENT, getCallOptions(), request);
+        }
+    }
+
+    /**
+     *
+     */
+    public static final class CallbackServiceFutureStub extends io.grpc.stub.AbstractStub<CallbackServiceFutureStub> {
+
+        private CallbackServiceFutureStub(io.grpc.Channel channel) {
+            super(channel);
+        }
+
+        private CallbackServiceFutureStub(io.grpc.Channel channel,
+            io.grpc.CallOptions callOptions) {
+            super(channel, callOptions);
+        }
+
+        @Override
+        protected CallbackServiceFutureStub build(io.grpc.Channel channel,
+            io.grpc.CallOptions callOptions) {
+            return new CallbackServiceFutureStub(channel, callOptions);
+        }
+
+        /**
+         * <pre>
+         * Subscribes events
+         * </pre>
+         */
+        public com.google.common.util.concurrent.ListenableFuture<org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent> onTopicEvent(
+            org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent request) {
+            return futureUnaryCall(
+                getChannel().newCall(METHOD_ON_TOPIC_EVENT, getCallOptions()), request);
+        }
+    }
+
+    private static final int METHODID_ON_TOPIC_EVENT = 0;
+
+    private static final class MethodHandlers<Req, Resp> implements
+        io.grpc.stub.ServerCalls.UnaryMethod<Req, Resp>,
+        io.grpc.stub.ServerCalls.ServerStreamingMethod<Req, Resp>,
+        io.grpc.stub.ServerCalls.ClientStreamingMethod<Req, Resp>,
+        io.grpc.stub.ServerCalls.BidiStreamingMethod<Req, Resp> {
+
+        private final CallbackServiceImplBase serviceImpl;
+        private final int methodId;
+
+        MethodHandlers(CallbackServiceImplBase serviceImpl, int methodId) {
+            this.serviceImpl = serviceImpl;
+            this.methodId = methodId;
+        }
+
+        @Override
+        @SuppressWarnings("unchecked")
+        public void invoke(Req request, io.grpc.stub.StreamObserver<Resp> responseObserver) {
+            switch (methodId) {
+                case METHODID_ON_TOPIC_EVENT:
+                    serviceImpl.onTopicEvent((org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent) request,
+                        (io.grpc.stub.StreamObserver<org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent>) responseObserver);
+                    break;
+                default:
+                    throw new AssertionError();
+            }
+        }
+
+        @Override
+        @SuppressWarnings("unchecked")
+        public io.grpc.stub.StreamObserver<Req> invoke(
+            io.grpc.stub.StreamObserver<Resp> responseObserver) {
+            switch (methodId) {
+                default:
+                    throw new AssertionError();
+            }
+        }
+    }
+
+    private static final class CallbackServiceDescriptorSupplier implements io.grpc.protobuf.ProtoFileDescriptorSupplier {
+
+        @Override
+        public com.google.protobuf.Descriptors.FileDescriptor getFileDescriptor() {
+            return EventMeshGrpcService.getDescriptor();
+        }
+    }
+
+    private static volatile io.grpc.ServiceDescriptor serviceDescriptor;
+
+    public static io.grpc.ServiceDescriptor getServiceDescriptor() {
+        io.grpc.ServiceDescriptor result = serviceDescriptor;
+        if (result == null) {
+            synchronized (CallbackServiceGrpc.class) {
+                result = serviceDescriptor;
+                if (result == null) {
+                    serviceDescriptor = result = io.grpc.ServiceDescriptor.newBuilder(SERVICE_NAME)
+                        .setSchemaDescriptor(new CallbackServiceDescriptorSupplier())
+                        .addMethod(METHOD_ON_TOPIC_EVENT)
+                        .build();
+                }
+            }
+        }
+        return result;
+    }
+}

eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/client/CloudEventsPublishInstance.java

@@ -0,0 +1,89 @@
+/*
+ * 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.eventmesh.connector.openfunction.client;
+
+import org.apache.eventmesh.client.grpc.config.EventMeshGrpcClientConfig;
+import org.apache.eventmesh.client.grpc.producer.EventMeshGrpcProducer;
+import org.apache.eventmesh.client.tcp.common.EventMeshCommon;
+import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.common.utils.ThreadUtils;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class CloudEventsPublishInstance {
+
+    // This messageSize is also used in SubService.java (Subscriber)
+    public static final int MESSAGE_SIZE = 5;
+
+    public static void main(String[] args) throws Exception {
+
+        try (EventMeshGrpcProducer eventMeshGrpcProducer = new EventMeshGrpcProducer(
+            initEventMeshGrpcClientConfig("FUNCTION_PRODUCER_GROUP"))) {
+
+            final Map<String, String> content = new HashMap<>();
+            content.put("content", "testAsyncMessage");
+
+            for (int i = 0; i < MESSAGE_SIZE; i++) {
+                eventMeshGrpcProducer.publish(buildCloudEvent(content));
+                ThreadUtils.sleep(1, TimeUnit.SECONDS);
+            }
+
+            ThreadUtils.sleep(30, TimeUnit.SECONDS);
+        }
+    }
+
+    protected static EventMeshGrpcClientConfig initEventMeshGrpcClientConfig(final String groupName) throws IOException {
+        final String eventMeshIp = "127.0.0.1";
+        final String eventMeshGrpcPort = "10110";
+
+        return EventMeshGrpcClientConfig.builder()
+            .serverAddr(eventMeshIp)
+            .serverPort(Integer.parseInt(eventMeshGrpcPort))
+            .producerGroup(groupName)
+            .env("env")
+            .idc("idc")
+            .sys("1234")
+            .build();
+    }
+
+    protected static CloudEvent buildCloudEvent(final Map<String, String> content) {
+        return CloudEventBuilder.v1()
+            .withId(UUID.randomUUID().toString())
+            .withSubject("TEST-TOPIC-FUNCTION")
+            .withSource(URI.create("/"))
+            .withDataContentType("application/cloudevents+json")
+            .withType(EventMeshCommon.CLOUD_EVENTS_PROTOCOL_NAME)
+            .withData(JsonUtils.toJSONString(content).getBytes(StandardCharsets.UTF_8))
+            .withExtension(Constants.EVENTMESH_MESSAGE_CONST_TTL, String.valueOf(4 * 1000))
+            .build();
+
+    }
+}

eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/client/CloudEventsSubscribeInstance.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.eventmesh.connector.openfunction.client;
+
+import io.grpc.Server;
+import io.grpc.ServerBuilder;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class CloudEventsSubscribeInstance {
+
+    public static void main(String[] args) throws Exception {
+        Server server = ServerBuilder.forPort(10115).addService(new CallbackService()).build();
+        server.start();
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            try {
+                server.shutdown();
+            } catch (Exception e) {
+                log.error("exception when shutdown.", e);
+            }
+        }));
+        server.awaitTermination();
+    }
+}

eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/client/EventMeshGrpcService.java

@@ -0,0 +1,79 @@
+/*
+ * 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.
+ */
+
+// Generated by the protocol buffer compiler.  DO NOT EDIT!
+// source: callback-service.proto
+
+package org.apache.eventmesh.connector.openfunction.client;
+
+public final class EventMeshGrpcService {
+
+    private EventMeshGrpcService() {
+    }
+
+    public static void registerAllExtensions(
+        com.google.protobuf.ExtensionRegistryLite registry) {
+    }
+
+    public static void registerAllExtensions(
+        com.google.protobuf.ExtensionRegistry registry) {
+        registerAllExtensions(
+            (com.google.protobuf.ExtensionRegistryLite) registry);
+    }
+
+    public static com.google.protobuf.Descriptors.FileDescriptor
+    getDescriptor() {
+        return descriptor;
+    }
+
+    private static com.google.protobuf.Descriptors.FileDescriptor
+        descriptor;
+
+    static {
+        String[] descriptorData = {
+            "\n\026callback-service.proto\022#org.apache.eve" +
+                "ntmesh.cloudevents.v1\032\033eventmesh-cloudev" +
+                "ents.proto\032\031google/protobuf/any.proto\032\037g" +
+                "oogle/protobuf/timestamp.proto2\203\001\n\017Callb" +
+                "ackService\022p\n\014OnTopicEvent\022/.org.apache." +
+                "eventmesh.cloudevents.v1.CloudEvent\032/.or" +
+                "g.apache.eventmesh.cloudevents.v1.CloudE" +
+                "ventBA\n)org.apache.eventmesh.connect.ope" +
+                "nfunctionB\024EventMeshGrpcServiceb\006proto3"
+        };
+        com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
+            new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
+                public com.google.protobuf.ExtensionRegistry assignDescriptors(
+                    com.google.protobuf.Descriptors.FileDescriptor root) {
+                    descriptor = root;
+                    return null;
+                }
+            };
+        com.google.protobuf.Descriptors.FileDescriptor
+            .internalBuildGeneratedFileFrom(descriptorData,
+                new com.google.protobuf.Descriptors.FileDescriptor[] {
+                    org.apache.eventmesh.common.protocol.grpc.cloudevents.EventMeshCloudEvents.getDescriptor(),
+                    com.google.protobuf.AnyProto.getDescriptor(),
+                    com.google.protobuf.TimestampProto.getDescriptor(),
+                }, assigner);
+        org.apache.eventmesh.common.protocol.grpc.cloudevents.EventMeshCloudEvents.getDescriptor();
+        com.google.protobuf.AnyProto.getDescriptor();
+        com.google.protobuf.TimestampProto.getDescriptor();
+    }
+
+    // @@protoc_insertion_point(outer_class_scope)
+}

eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/config/OpenFunctionServerConfig.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.eventmesh.connector.openfunction.config;
+
+import org.apache.eventmesh.openconnect.api.config.Config;
+
+import lombok.Data;
+
+@Data
+public class OpenFunctionServerConfig extends Config {
+
+    private int serverPort;
+
+    private boolean sourceEnable;
+
+    private boolean sinkEnable;
+
+    private String targetAddress;
+
+    private int targetPort;
+
+}