[Feature #4793] Support MQTT protocol #4794

karsonto · 2024-03-19T09:55:19Z

Fixes #4793

Motivation

Explain the content here.
Explain why you want to make the changes and what problem you're trying to solve.

Modifications

Describe the modifications you've done.

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
If a feature is not applicable for documentation, explain why?
If a feature is not documented yet in this PR, please create a followup issue for adding the documentation

…eadUtils] (apache#4297) * Update ThreadUtils.java * Update ThreadUtils.java

…ectors] (apache#4314) * [ISSUE apache#4266]: Retore interrupted state for interrupted exception in open function sinc connector * [ISSUE apache#4266]: Retore interrupted state for interrupted exception in open function source connector * [ISSUE apache#4266]: Retore interrupted state for interrupted exception in RocketMQSink connector

…pache#4315) * [ISSUE apache#4093]: Refactor package building chunks to reusable method * [ISSUE apache#4093]: Refactor user agent building chunks to reusable method

…aced with lambda[HelloTask] (apache#4316) * [4260] Fixed: Anonymous new ChannelFutureListener() can be replaced with lambda[HelloTask] * Upgraded: First Interaction Action Version to Latest * [4260] Removed: Unused Import * Update package import order in HelloTask.java --------- Co-authored-by: mike_xwm <[email protected]>

* fix: add cloudEventCodec for redis connector. * fix: add cloudEventCodec for redis connector. * fix import order. * fix: add doc. * fix: adjust cloudevent encode and decode.

…-offsetStorage

…che#4337) * issue-4262 Enhancement Request EventMeshCloudEventUtils * issue-4262 Enhancement Request EventMeshCloudEventUtils move on top method getValue * issue-4262 Enhancement Request EventMeshCloudEventUtils remove redundant else for Optional.of --------- Co-authored-by: maxim.zgardan <[email protected]>

[ISSUE apache#4328] Add offsetManagement Service for connectors

[ISSUE apache#4268] Used switch to replace the if-else [CloudEventsProtocolAdaptor]

[ISSUE apache#4283] Add event-bridge example

… test coverage rate (apache#4340) * issues apache#4264 commit * unit * issues apache#4339 unit test coverage * issues apache#4339 WebhookProtocolTransportObjectTest fix * issues apache#4339 WebhookProtocolTransportObjectTest fix * issues apache#4339 WebhookProtocolTransportObjectTest fix * issues apache#4339 WebhookProtocolTransportObjectTest fix * issues apache#4339 check style * issues apache#4339 new instance replace builder. * issues apache#4339 WebhookProtocolTransportObjectTest unit test. * issues apache#4339 check style. * issues apache#4339 check style. * issues apache#4339 check style. * fixed. * fixed. * fixed. * fixed. * fixed. * delete java bean unit test. * fix build error.

* fix start error and some code optimization. * fix code style

…ache#4348) * [ISSUE apache#4339][Unit Test] eventmesh-common header unit test. * [ISSUE apache#4339][Unit Test] eventmesh-common header unit test. * [ISSUE apache#4339][Unit Test] eventmesh-common body unit test. * [ISSUE apache#4339][Unit Test] eventmesh-common ReflectUtils and RandomStringUtils unit test. * [ISSUE apache#4339]comment. * [ISSUE apache#4339]expectedException.

…che#4356)

…hrow java.lang.NullPointerException (apache#4352) * [ISSUE apache#4345]Fix publish EventMeshMessage without requestCode throw java.lang.NullPointerException * optimize logic

…he#4762) Co-authored-by: orol116 <[email protected]>

…y server (apache#4739) * fix 4738 * fix some bug * fix bug * remove initProducerManager from AbstractRemotingServer init * bug fix * bug fix * some enhance * some enhance * add admin bootstrap * some enhance * remove HttpHandlerManager and ClientManageController. * modify some unit test * add admin http handlermanager

* [ISSUE apache#4458] Support mysql Sink Connector feature * remove pg jdbc import * update dependencies

* resolve conflicts. * resolve

…_CHARSET (apache#4776)

…ache#4777) * Add null check in writeOffset method * delete todo * Move data.put inside null check in writeOffset method * simplify if judgement * remove dev environment * fix style

… when dequeue (apache#4779) * bug fix * bug fix

a. Change to private modifier. b. Repeat code extraction as method.

…tUtils

* condition_check_for_source_eorker * ci_typo_fix * import_order_fix

codecov · 2024-03-19T10:21:23Z

Codecov Report

Attention: Patch coverage is 38.42207% with 960 lines in your changes missing coverage. Please review.

Project coverage is 16.11%. Comparing base (1600c6b) to head (44e61e9).
Report is 4687 commits behind head on master.

Files with missing lines	Patch %	Lines
...mesh/connector/jdbc/connection/JdbcConnection.java	0.00%	137 Missing ⚠️
...nnector/file/sink/connector/FileSinkConnector.java	0.00%	62 Missing ⚠️
...r/jdbc/dialect/AbstractGeneralDatabaseDialect.java	0.00%	56 Missing ⚠️
...g/apache/eventmesh/common/utils/JsonPathUtils.java	0.00%	54 Missing ⚠️
...ava/org/apache/eventmesh/common/utils/LogUtil.java	7.54%	48 Missing and 1 partial ⚠️
...tor/jdbc/connection/mysql/MysqlJdbcConnection.java	0.00%	43 Missing ⚠️
...a/org/apache/eventmesh/connector/jdbc/Payload.java	0.00%	32 Missing ⚠️
...dingtalk/sink/connector/DingDingSinkConnector.java	65.38%	24 Missing and 3 partials ⚠️
...pache/eventmesh/connector/jdbc/CatalogChanges.java	0.00%	26 Missing ⚠️
...ava/org/apache/eventmesh/connector/jdbc/Field.java	0.00%	24 Missing ⚠️
... and 82 more

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #4794      +/-   ##
============================================
- Coverage     16.93%   16.11%   -0.82%     
- Complexity     1413     1734     +321     
============================================
  Files           589      870     +281     
  Lines         25786    31682    +5896     
  Branches       2397     2739     +342     
============================================
+ Hits           4367     5106     +739     
- Misses        20983    26097    +5114     
- Partials        436      479      +43

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

pandaapo · 2024-03-19T10:16:57Z

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshMQTTServer.java

+
+
+    protected final transient Map<MqttMessageType, MqttProcessor> processorTable =
+        new ConcurrentHashMap<>(64);


The class is not serializable, is the transient keyword redundant?

Noted，please review again.

pandaapo · 2024-03-19T10:24:54Z

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshMQTTServer.java

+                } catch (Exception ex) {
+                    log.error("EventMeshMQTTServer RemotingServer shutdown Err!", ex);
+                }
+                System.exit(-1);


Is it appropriate to exit the process when the MQTT server fails to start?

This is a reference to other EM protocols.If you want to modify it after the discussion, I will ignore this startup failure.

@xwm1992

Currently tcp protocol, http protocol server must be started successfully, grpc protocol did not so. Now added MQTT protocol server must be started successfullyt, please community to give advice, I can not decide.

目前tcp协议、http协议的server必须成功启动，grpc协议没有如此，现在新增MQTT协议的server是否必须启动成功，请社区给出意见，我权衡不好。

我理解这里的必须启动成功实际上是受是否开启MQTT协议的配置控制的吧？这里的退出我认为没有问题，如果MQTT协议加载有问题退出了，那其实可以在配置中移除MQTT协议，保证TCP、HTTP等协议正常启动服务就好。

pandaapo · 2024-03-19T10:43:29Z

...time/src/main/java/org/apache/eventmesh/runtime/core/protocol/mqtt/client/ClientManager.java

+        this.cleanThread = new Thread(() -> {
+            while (true && !Thread.currentThread().isInterrupted()) {
+                try {
+                    ClientInfo clientInfo = delayQueue.take();


What is the role of delayQueue?

The role of delayQueue is to clean up the time out connection.

Is it up to the client to decide and set the length of time that each client's connection survives when it sends a message?

每个客户端的连接的存活时长，是靠客户端发送消息时自己决定并设置好吗？

客户端可以携带keep alive time参数，用做会话存活时间

pandaapo · 2024-03-19T10:44:35Z

...rc/main/java/org/apache/eventmesh/runtime/core/protocol/mqtt/processor/PublishProcessor.java

+
+    @Override
+    public void process(ChannelHandlerContext ctx, MqttMessage mqttMessage) throws MqttException {
+        //


Remove this.

Noted，please review again.

pandaapo · 2024-03-21T11:19:35Z

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshMQTTServer.java

+    protected final Map<MqttMessageType, MqttProcessor> processorTable =
+        new ConcurrentHashMap<>(64);
+
+    private final transient AtomicBoolean started = new AtomicBoolean(false);


Same problem.
I'm not familiar with MQTT, so the rest of review work needs the community to complete.

Same problem.

Waiting for your response.

what's the problem here？

#4794 (comment)

Noted，please review again.

xwm1992 · 2024-03-26T02:51:12Z

@karsonto
I don’t understand the implementation of the publish processor, and I don’t see how messages of the mqtt protocol are converted and sent to eventmesh storage.

我没有理解对于publish processor的实现，没有看到将mqtt协议的消息进行转换后发送到eventmesh storage。

karsonto · 2024-03-26T02:54:38Z

@xwm1992
This PR is the framework design of the MQTT protocol , and will continue to optimize persist messages in another PR.

karsonto · 2024-03-26T07:33:48Z

This PR is the framework design of the MQTT protocol , and will continue to optimize persist messages in another PR.

Pil0tXia

Is eventmesh-protocol-plugin implementation the only work left for this protocol in Runtime?

karsonto · 2024-04-15T01:30:14Z

Is eventmesh-protocol-plugin implementation the only work left for this protocol in Runtime?

I think there is unnecessry to implement this module for this time. If it needs to be implemented in the future, I will submit a PR to supplement it.

github-actions · 2024-08-11T18:34:32Z

It has been 60 days since the last activity on this pull request. I am reaching out here to gently remind you that the Apache EventMesh community values every pull request, and please feel free to get in touch with the reviewers at any time. They are available to assist you in advancing the progress of your pull request and offering the latest feedback.

If you encounter any challenges during development, seeking support within the community is encouraged. We sincerely appreciate your contributions to Apache EventMesh.

github-actions · 2025-04-14T18:38:01Z

It has been 60 days since the last activity on this pull request. I am reaching out here to gently remind you that the Apache EventMesh community values every pull request, and please feel free to get in touch with the reviewers at any time. They are available to assist you in advancing the progress of your pull request and offering the latest feedback.

If you encounter any challenges during development, seeking support within the community is encouraged. We sincerely appreciate your contributions to Apache EventMesh.

qqeasonchen and others added 30 commits August 1, 2023 11:16

Update enhancement_request.yml (apache#4312)

7fd6e3d

Update feature_request.yml (apache#4313)

3592263

[ISSUE apache#4171]support pulsar connector (apache#4186)

dd40227

[ISSUE apache#4263] InterruptedExceptions should never be ignored[Thr…

dff6e7a

…eadUtils] (apache#4297) * Update ThreadUtils.java * Update ThreadUtils.java

[ISSUE apache#4093]Repeat code extraction as method.[MessageUtils] (a…

99bcf67

…pache#4315) * [ISSUE apache#4093]: Refactor package building chunks to reusable method * [ISSUE apache#4093]: Refactor user agent building chunks to reusable method

modify the offset management service

4121332

[Feature apache#4032] Fix redis codec (apache#4301)

19dd2b1

* fix: add cloudEventCodec for redis connector. * fix: add cloudEventCodec for redis connector. * fix import order. * fix: add doc. * fix: adjust cloudevent encode and decode.

modify the nacos config management service

939db45

modify the configs

8ab4000

[ISSUE apache#4264]Some optimizations for ProducerService (apache#4334)

800b628

Merge branch 'master' of https://github.com/apache/eventmesh into add…

4a12417

…-offsetStorage

fix ci check error

4f612f1

fix ci check error

4eddf36

fix ci check error

9621900

[ISSUE apache#4085] Switching order (apache#4338)

9d67039

Merge pull request apache#4329 from xwm1992/add-offsetStorage

50bbaec

[ISSUE apache#4328] Add offsetManagement Service for connectors

fix template not taking effect (apache#4325)

97157f8

Merge pull request apache#4298 from devCod3r/developmentEM

8732dd3

[ISSUE apache#4268] Used switch to replace the if-else [CloudEventsProtocolAdaptor]

Merge pull request apache#4284 from seriouszyx/yixiang/event-bridge

4568c50

[ISSUE apache#4283] Add event-bridge example

feat: refactor with mongodb plugin.

475cb2c

[ISSUE 4346] Fix start error and some code optimization. (apache#4347)

dec34f3

* fix start error and some code optimization. * fix code style

[ISSUE 4350] Fixed eventmesh startup error under MQ plugin (apache#4351)

8418745

[ISSUE apache#4354] Fix spelling mistake (apache#4353)

dbfb571

[ISSUE apache#4191] Do some code optimization.[StreamPushRequest](apa…

85ee0fb

…che#4356)

[ISSUE apache#4345]Fix publish EventMeshMessage without requestCode t…

8be4180

…hrow java.lang.NullPointerException (apache#4352) * [ISSUE apache#4345]Fix publish EventMeshMessage without requestCode throw java.lang.NullPointerException * optimize logic

orol116 and others added 12 commits February 1, 2024 23:43

[ISSUE apache#4761] Update copyright year to 2024 (apache#4761) (apac…

28c746c

…he#4762) Co-authored-by: orol116 <[email protected]>

Update README.md (apache#4772)

36888b8

[ISSUE apache#4458] Support MySQL Sink Connector feature (apache#4771)

d6393ab

* [ISSUE apache#4458] Support mysql Sink Connector feature * remove pg jdbc import * update dependencies

[ISSUE apache#4731] HttpRequestProcessor enhancement (apache#4732)

9a3912a

* resolve conflicts. * resolve

[ISSUE apache#3014]Replace encoding using the static variable DEFAULT…

5cc065f

…_CHARSET (apache#4776)

[ISSUE apache#4770] Empty class member cause NullpointerException (ap…

1a5e06c

…ache#4777) * Add null check in writeOffset method * delete todo * Move data.put inside null check in writeOffset method * simplify if judgement * remove dev environment * fix style

[ISSUE apache#4778] Fix Standalone MessageQueue does not down counter…

c93fc3e

… when dequeue (apache#4779) * bug fix * bug fix

[ISSUE apache#3833] optimization ZookeeperMetaService (apache#4780)

713f7c6

a. Change to private modifier. b. Repeat code extraction as method.

[ISSUE apache#3828] Code optimization for StandaloneAdminTest and Tes…

c700a64

…tUtils

[ISSUE apache#4785] Condition check for source worker (apache#4786)

0250a89

* condition_check_for_source_eorker * ci_typo_fix * import_order_fix

init mqtt function

d0b3678

pandaapo reviewed Mar 19, 2024

View reviewed changes

some enhance.

1f8cf13

pandaapo reviewed Mar 21, 2024

View reviewed changes

some enhance.

44e61e9

Pil0tXia reviewed Apr 14, 2024

View reviewed changes

Pil0tXia added the need review The review work of this PR needs another reviewer/PMC's help label Jun 12, 2024

github-actions bot added the Stale label Aug 11, 2024

xwm1992 force-pushed the master branch from 4ea9f76 to e8fdcd3 Compare February 13, 2025 03:23

github-actions bot removed the Stale label Feb 13, 2025

github-actions bot added Stale and removed Stale labels Apr 14, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Feature #4793] Support MQTT protocol #4794

[Feature #4793] Support MQTT protocol #4794

karsonto commented Mar 19, 2024

codecov bot commented Mar 19, 2024 •

edited

Loading

pandaapo Mar 19, 2024

karsonto Mar 21, 2024

pandaapo Mar 19, 2024

karsonto Mar 21, 2024

pandaapo Mar 21, 2024

xwm1992 Mar 26, 2024

pandaapo Mar 19, 2024

karsonto Mar 21, 2024

pandaapo Mar 26, 2024

karsonto Mar 26, 2024

pandaapo Mar 19, 2024

karsonto Mar 21, 2024

pandaapo Mar 21, 2024

pandaapo Mar 26, 2024

karsonto Mar 26, 2024

pandaapo Mar 27, 2024

karsonto Mar 28, 2024

xwm1992 commented Mar 26, 2024

karsonto commented Mar 26, 2024 •

edited

Loading

karsonto commented Mar 26, 2024

Pil0tXia left a comment

karsonto commented Apr 15, 2024 •

edited

Loading

github-actions bot commented Aug 11, 2024

github-actions bot commented Apr 14, 2025



		protected final transient Map<MqttMessageType, MqttProcessor> processorTable =
		new ConcurrentHashMap<>(64);

[Feature #4793] Support MQTT protocol #4794

Are you sure you want to change the base?

[Feature #4793] Support MQTT protocol #4794

Conversation

karsonto commented Mar 19, 2024

Motivation

Modifications

Documentation

codecov bot commented Mar 19, 2024 • 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

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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xwm1992 commented Mar 26, 2024

karsonto commented Mar 26, 2024 • edited Loading

karsonto commented Mar 26, 2024

Pil0tXia left a comment

Choose a reason for hiding this comment

karsonto commented Apr 15, 2024 • edited Loading

github-actions bot commented Aug 11, 2024

github-actions bot commented Apr 14, 2025

codecov bot commented Mar 19, 2024 •

edited

Loading

karsonto commented Mar 26, 2024 •

edited

Loading

karsonto commented Apr 15, 2024 •

edited

Loading