[ISSUE #790] Knative Connector: Initial Implementation of Producer. #1027

pchengma · 2022-07-16T02:58:55Z

Fixing ISSUE #790.

Motivation

Implemented and tested Producer module to publish an event to Knative (ISSUE #790).
The PRs for ISSUE #790 will be merged into knative-connector branch.

Modifications

Followed the API design style of eventmesh-connector-api and made the following modifications.

Implementation

Producer for publishing an event message to Knative
CloudEvent Reader, Writer, and Header
Connector
Unit test for Producer
Included eventmesh-connector-knative in setting.gradle

How to Test

Assume the cloudevents-player Knative service has already been set up on your local machine; otherwise, please follow the steps in https://knative.dev/docs/getting-started/first-source/#sending-an-event to set up a Knative service (cloudevents-player) as a sink.

# Set eventMesh.connector.plugin.type=knative in eventmesh-runtime/conf/eventmesh.properties
# Do unit test:
$ cd incubator-eventmesh/eventmesh-connector-plugin/eventmesh-connector-knative
$ ../../gradlew clean test --tests KnativeConnectorTest.testConnect
# See the delivered event:
$ curl http://cloudevents-player.default.127.0.0.1.sslip.io/messages

Documentation

Does this pull request introduce a new feature? yes.
If yes, how is the feature documented? not documented.
If a feature is not documented yet in this PR, please create a followup issue for adding the documentation. The followup issues or pull requests for documentation will be created after almost all implementation work has been finished.

github-actions

Welcome to the Apache EventMesh (incubating) community!!
This is your first PR in our project. We're very excited to have you onboard contributing. Your contributions are greatly appreciated!

Please make sure that the changes are covered by tests.
We will be here shortly.
Let us know if you need any help!

Want to get closer to the community?

WeChat Group：

Mailing Lists:

Name	Description	Subscribe	Unsubscribe	Archive
Users	User support and questions mailing list	Subscribe	Unsubscribe	Mail Archives
Development	Development related discussions	Subscribe	Unsubscribe	Mail Archives
Commits	All commits to repositories	Subscribe	Unsubscribe	Mail Archives

pchengma · 2022-07-16T03:00:02Z

@xwm1992 Please have a look when you have time. Thanks very much in advance!

codecov · 2022-07-19T02:37:20Z

Codecov Report

Merging #1027 (14810a0) into knative-connector (72a7b33) will increase coverage by 0.17%.
The diff coverage is 52.98%.

❗ Current head 14810a0 differs from pull request most recent head f1e9245. Consider uploading reports for the commit f1e9245 to get more accurate results

@@                  Coverage Diff                   @@
##             knative-connector   #1027      +/-   ##
======================================================
+ Coverage                 9.69%   9.87%   +0.17%     
- Complexity                 610     629      +19     
======================================================
  Files                      362     374      +12     
  Lines                    23118   23244     +126     
  Branches                  2546    2541       -5     
======================================================
+ Hits                      2241    2295      +54     
- Misses                   20675   20740      +65     
- Partials                   202     209       +7

Impacted Files	Coverage Δ
...ve/cloudevent/impl/KnativeBinaryMessageReader.java	`0.00% <0.00%> (ø)`
...nector/knative/cloudevent/impl/KnativeHeaders.java	`0.00% <0.00%> (ø)`
...h/connector/knative/common/EventMeshConstants.java	`0.00% <0.00%> (ø)`
...tmesh/connector/knative/utils/CloudEventUtils.java	`0.00% <0.00%> (ø)`
...onnector/knative/producer/KnativeProducerImpl.java	`47.05% <47.05%> (ø)`
...tmesh/connector/knative/producer/ProducerImpl.java	`50.00% <50.00%> (ø)`
...connector/knative/config/ConfigurationWrapper.java	`52.63% <52.63%> (ø)`
.../knative/cloudevent/impl/KnativeMessageWriter.java	`66.66% <66.66%> (ø)`
...h/connector/knative/producer/AbstractProducer.java	`66.66% <66.66%> (ø)`
.../connector/knative/config/ClientConfiguration.java	`90.00% <90.00%> (ø)`
... and 10 more

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

xwm1992 · 2022-07-19T02:42:20Z

...ative/src/main/java/org/apache/eventmesh/connector/knative/producer/KnativeProducerImpl.java

+    }
+
+    @Override
+    public void sendOneway(CloudEvent cloudEvent) {


Please not implement this method, you should impelement the publish method with the send call back.

Thanks for your comments. I will change to the publish method instead.

Implemented publish method with SendCallback.

xwm1992 · 2022-07-19T02:42:31Z

...-knative/src/main/java/org/apache/eventmesh/connector/knative/producer/AbstractProducer.java

+        // Set HTTP header for CloudEvent:
+        this.httpUrlConnection.setRequestProperty(KnativeHeaders.CONTENT_TYPE, properties.getProperty(KnativeHeaders.CONTENT_TYPE));
+        this.httpUrlConnection.setRequestProperty(KnativeHeaders.CE_ID, properties.getProperty(KnativeHeaders.CE_ID));
+        this.httpUrlConnection.setRequestProperty(KnativeHeaders.CE_SPECVERSION, properties.getProperty(KnativeHeaders.CE_SPECVERSION));
+        this.httpUrlConnection.setRequestProperty(KnativeHeaders.CE_TYPE, properties.getProperty(KnativeHeaders.CE_TYPE));
+        this.httpUrlConnection.setRequestProperty(KnativeHeaders.CE_SOURCE, properties.getProperty(KnativeHeaders.CE_SOURCE));


These code for HTTP header attributes with the request, I recommend that you can write them when you send the cloud events not when you init the producer`.

Thanks for your comments. I will move this part out of the init period.

Moved HTTP header setting into send method.

xwm1992 · 2022-07-19T02:57:18Z

...ctor-knative/src/main/java/org/apache/eventmesh/connector/knative/producer/ProducerImpl.java

+    public void sendOneway(CloudEvent cloudEvent) {
+        // Get CloudEvent data:
+        try {
+            String data = KnativeMessageFactory.createReader(cloudEvent);
+            super.getHttpUrlConnection().getOutputStream().write(data.getBytes(StandardCharsets.UTF_8));
+
+            // Send CloudEvent message:
+            String s = "";
+            int code = super.getHttpUrlConnection().getResponseCode();
+            if (code == 200) {
+                BufferedReader reader = new BufferedReader(
+                        new InputStreamReader(super.getHttpUrlConnection().getInputStream()));
+                String line;
+                while ((line = reader.readLine()) != null) {
+                    s += line + "\n";
+                }
+                reader.close();
+            }
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+}


You don't need to have this method, and when you implement the publish method with the callback, you can use the Async Http Client, here don't implement synchronously.

Thanks for your comments. I will change to the asynchronous way to call the publish method.

Implemented sendAsync method by using AsyncHttpClient.

Added a simple exception checker method checkProducerException and a utility method convertSendResult to help implement methods with SendCallback.

Added AsyncHttpClient dependency to build.gradle file.

xwm1992 · 2022-07-19T03:00:57Z

...c/main/java/org/apache/eventmesh/connector/knative/cloudevent/impl/KnativeMessageWriter.java

+    public KnativeMessageWriter(String data) {
+        String s = "{ \"msg\": [\"" + data + "\"]}";
+        this.message = new CloudEventBuilder()
+                .withId("my-id")
+                .withSource(URI.create("/myClient"))
+                .withType("dev.knative.cronjob.event")
+                .withDataContentType("application/json")
+                .withData(s.getBytes(StandardCharsets.UTF_8))
+                .build();
+    }


These codes look strange, you use the cloudeventbuilder to build the cloud events, but you only want to put the data, other attributes are useless.

Thanks for your comments. I will remove the unnecessary attributes.

Removed unnecessary attributes in KnativeMessageWriter.

1. Changed to load Knative service URL from properties file (added common/EventMeshConstants, config/ClientConfiguration, and config/ConfigurationWrapper). 2. Deleted unused sendCallbackConvert method. 3. Added exception handling for HTTP response code error in send method. 4. Added related build.gradle dependencies, tools/known-dependencies.txt dependency, and third-party-licenses/LICENSE-httpasyncclient.txt file.

1. Fixed issue in Continuous Integration / Build (ubuntu-latest, 8) (pull_request). 2. Fixed issue in Continuous Integration / Build (ubuntu-latest, 11) (pull_request). 3. Fixed issue in Continuous Integration / Build (macOS-latest, 8) (pull_request). 4. Fixed issue in Continuous Integration / Build (macOS-latest, 11) (pull_request). 5. Fixed issue in Continuous Integration / License Check (pull_request).

Uploaded implementation of Knative producer.

6e8c485

github-actions bot reviewed Jul 16, 2022

View reviewed changes

Uploaded implementation of Knative producer.

6cd8f02

xwm1992 requested changes Jul 19, 2022

View reviewed changes

pchengma added 2 commits July 20, 2022 00:24

Amended Knative producer implementation according to comments.

93c9a38

Amended Knative producer implementation according to comments.

d6a939a

pchengma requested a review from xwm1992 July 20, 2022 01:24

pchengma added 8 commits July 21, 2022 10:45

Amended Knative producer implementation according to comments.

a854445

Fixed checkstyle and license check issues.

a7e1e49

Passed producer test for the whole pub/sub procedure.

079e246

Minor fixed for checkstyle issue.

7d7234a

Added start and shutdown methods.

ba9cd06

Added license in KnativeConnectorTest.

f1e9245

qqeasonchen merged commit be661a3 into apache:knative-connector Aug 26, 2022

pchengma mentioned this pull request Sep 1, 2022

[ISSUE #790] Knative Connector: Merge Request from knative-connector to master. #1214

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[ISSUE #790] Knative Connector: Initial Implementation of Producer. #1027

[ISSUE #790] Knative Connector: Initial Implementation of Producer. #1027

pchengma commented Jul 16, 2022 •

edited

Loading

github-actions bot left a comment

pchengma commented Jul 16, 2022

codecov bot commented Jul 19, 2022 •

edited

Loading

xwm1992 Jul 19, 2022

pchengma Jul 19, 2022

pchengma Jul 19, 2022

xwm1992 Jul 19, 2022

pchengma Jul 19, 2022

pchengma Jul 19, 2022 •

edited

Loading

xwm1992 Jul 19, 2022

pchengma Jul 19, 2022

pchengma Jul 19, 2022 •

edited

Loading

xwm1992 Jul 19, 2022

pchengma Jul 19, 2022

pchengma Jul 19, 2022

[ISSUE #790] Knative Connector: Initial Implementation of Producer. #1027

[ISSUE #790] Knative Connector: Initial Implementation of Producer. #1027

Conversation

pchengma commented Jul 16, 2022 • edited Loading

Motivation

Modifications

Implementation

How to Test

Documentation

github-actions bot left a comment

Choose a reason for hiding this comment

pchengma commented Jul 16, 2022

codecov bot commented Jul 19, 2022 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pchengma Jul 19, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pchengma Jul 19, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pchengma commented Jul 16, 2022 •

edited

Loading

codecov bot commented Jul 19, 2022 •

edited

Loading

pchengma Jul 19, 2022 •

edited

Loading

pchengma Jul 19, 2022 •

edited

Loading