Virtual transport of RocketMQ by imzs · Pull Request #2306 · celery/kombu

imzs · 2025-05-30T07:19:13Z

#2305
Add virtual transport for RocketMQ

auvipy

can you please also fix the failing test

=================================== FAILURES ===================================
____________________________ test_Channel.test_get _____________________________

self = <t.unit.transport.test_rocketmq.test_Channel object at 0x7f75eb63bbf0>

def test_get(self):
    queue = 'new-queue'
    self.channel.basic_consume(queue, True, None, 'cg1')

    self.consumer.receive.return_value = []
    with pytest.raises(Empty):
        self.channel._get(queue)

    rocketmq_message = _message_to_rocketmq_ack_message(_mock_message(topic=queue, queue=queue))

  rocketmq_message.body = str_to_bytes(dumps(self.channel.prepare_message({})))

t/unit/transport/test_rocketmq.py:150:

self = <rocketmq.v5.model.message.Message object at 0x7f75d9132c00>
body = b'{"body": {}, "content-encoding": null, "content-type": null, "headers": {}, "properties": {"delivery_info": {}, "priority": 0}}'
Restoring 2 unacknowledged message(s)

@body.setter
def body(self, body):

  if body is None or body.strip() == "":

E BytesWarning: Comparison between bytes and string

.tox/3.12-unit/lib/python3.12/site-packages/rocketmq/v5/model/message.py:175: BytesWarning

for more information, see https://pre-commit.ci

imzs · 2025-06-01T07:49:39Z

can you please also fix the failing test

=================================== FAILURES =================================== ____________________________ test_Channel.test_get _____________________________

self = <t.unit.transport.test_rocketmq.test_Channel object at 0x7f75eb63bbf0>
def test_get(self):
    queue = 'new-queue'
    self.channel.basic_consume(queue, True, None, 'cg1')

    self.consumer.receive.return_value = []
    with pytest.raises(Empty):
        self.channel._get(queue)

    rocketmq_message = _message_to_rocketmq_ack_message(_mock_message(topic=queue, queue=queue))
  rocketmq_message.body = str_to_bytes(dumps(self.channel.prepare_message({})))
t/unit/transport/test_rocketmq.py:150:

self = <rocketmq.v5.model.message.Message object at 0x7f75d9132c00> body = b'{"body": {}, "content-encoding": null, "content-type": null, "headers": {}, "properties": {"delivery_info": {}, "priority": 0}}' Restoring 2 unacknowledged message(s)
@body.setter
def body(self, body):
  if body is None or body.strip() == "":
E BytesWarning: Comparison between bytes and string

.tox/3.12-unit/lib/python3.12/site-packages/rocketmq/v5/model/message.py:175: BytesWarning

hi, I've suppressed the BytesWarning for those test cases, and I'll submit an issue to the SDK community to get it fixed.

auvipy · 2025-06-02T10:51:02Z

Ok, thanks

imzs · 2025-06-25T08:53:53Z

@auvipy hi, I've fixed the issue which causes test failure in py313, could you help me trigger the workflow?

auvipy · 2025-06-25T08:55:11Z

Sure

imzs · 2025-11-06T12:42:13Z

@auvipy hi, the conflict was fixed and I notice that python-version has changed, so could you help me trigger the workflow?

codecov · 2025-11-09T13:28:07Z

Codecov Report

❌ Patch coverage is 80.66465% with 64 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.32%. Comparing base (5ef76b7) to head (cca60a4).
✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
kombu/transport/rocketmq.py	80.66%	48 Missing and 16 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2306      +/-   ##
==========================================
- Coverage   82.37%   82.32%   -0.06%     
==========================================
  Files          79       80       +1     
  Lines       10116    10447     +331     
  Branches     1159     1197      +38     
==========================================
+ Hits         8333     8600     +267     
- Misses       1580     1628      +48     
- Partials      203      219      +16

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

Copilot

Pull Request Overview

This PR adds RocketMQ transport support to Kombu, enabling message queuing through Apache RocketMQ as a backend. The implementation follows the virtual transport pattern and provides a comprehensive mapping between AMQP semantics and RocketMQ concepts.

Implements RocketMQ transport module with Channel, QoS, and Transport classes
Adds comprehensive test coverage for the new RocketMQ transport
Includes documentation and dependency configuration for the RocketMQ integration

Reviewed Changes

Copilot reviewed 7 out of 8 changed files in this pull request and generated 7 comments.

Show a summary per file

File	Description
kombu/transport/rocketmq.py	Main implementation of RocketMQ transport with Channel, QoS, and Transport classes, including message handling, producer/consumer management, and AMQP-to-RocketMQ mapping
t/unit/transport/test_rocketmq.py	Comprehensive unit tests for RocketMQ transport covering QoS operations, channel operations, producer/consumer management, and message handling
requirements/extras/rocketmq.txt	Defines RocketMQ-specific dependencies (rocketmq-python-client and grpcio-tools)
requirements/test-ci.txt	Adds RocketMQ requirements to CI test dependencies
requirements/extras/gcpubsub.txt	Updates protobuf version specification
kombu/transport/init.py	Registers the RocketMQ transport in the transport registry
docs/reference/kombu.transport.rocketmq.rst	Adds Sphinx documentation for the RocketMQ transport module
docs/reference/index.rst	Adds RocketMQ transport to the documentation index

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Change-Id: Ib0fcf35f367c3292ab415bcf46cfd4105c57248a

imzs · 2025-11-11T09:48:32Z

@auvipy I've fixed the typo, indentation, deprecated method in resolve-conflict branch based on Copilot's suggestions.

imzs · 2025-11-24T03:01:12Z

@auvipy Hi, is there anything else I should improve or clarify? I’m happy to make any further adjustments needed.

imzs · 2026-01-12T03:47:34Z

hi @auvipy
Could you please trigger the test again? The test case failed due to a timeout issue, but it should pass upon retry.

ERROR: HTTP error 504 while getting https://github.com/celery/py-amqp/zipball/main
ERROR: Could not install requirement https://github.com/celery/py-amqp/zipball/main 
(from -r /home/runner/_work/kombu/kombu/requirements/dev.txt (line 1)) 
because of HTTP error 504 Server Error: Gateway Time-out for url: https://github.com/celery/py-amqp/zipball/main for URL https://github.com/celery/py-amqp/zipball/main

auvipy

please fix the merge conflicts

imzs · 2026-03-02T01:48:47Z

please fix the merge conflicts

The gcpubsub requirement conflict has been resolved.

Copilot

Pull request overview

Copilot reviewed 7 out of 8 changed files in this pull request and generated 5 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Copilot

Pull request overview

Copilot reviewed 8 out of 9 changed files in this pull request and generated 3 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Copilot

Pull request overview

Copilot reviewed 8 out of 8 changed files in this pull request and generated 9 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-03-25T14:12:21Z

+            if message.topic in self._auto_ack_topics:
+                _ack_rocketmq_message(consumer, message, False)
+            amqp_queue = self.connection._topic_to_queue[message.topic]
+            payload = _rocketmq_message_to_payload(message, consumer.consumer_group, amqp_queue)
+            if callback is not None:


_get_bulk assumes every received message topic exists in self.connection._topic_to_queue and indexes it with [...]. If a message arrives for a topic that is no longer registered (e.g., races with basic_cancel, or subscriptions shared across topics/groups), this will raise KeyError and break the drain loop. Consider using .get() and handling unknown topics explicitly (e.g., log + ack/requeue/skip) to keep the consumer robust.

Copilot · 2026-03-25T14:12:22Z

+        if message.topic in self._auto_ack_topics:
+            _ack_rocketmq_message(consumer, message, False)
+        amqp_queue = self.connection._topic_to_queue[message.topic]
+        return _rocketmq_message_to_payload(message, consumer.consumer_group, amqp_queue)


_get() has the same assumption as _get_bulk() when resolving amqp_queue via self.connection._topic_to_queue[message.topic]. A missing mapping will raise KeyError and surface as an unexpected exception instead of being treated as an unconsumed/undeliverable message. Consider guarding this lookup (e.g., .get()) and handling the unknown topic similarly to _get_bulk().

Copilot · 2026-03-25T14:12:22Z

+        if topic in self._bound_topics:
+            return consumer
+
+        def _subscribe():
+            consumer.subscribe(
+                topic,
+                FilterExpression(
+                    expression=group_config.filter_exp,
+                    filter_type=group_config.filter_type,
+                ),
+            )
+            self._bound_topics.add(topic)
+
+        retry_over_time(
+            _subscribe, Exception, max_retries=4, interval_max=4, timeout=20
+        )
+        return consumer


_get_consumer(..., passive=True) currently prevents creating a new consumer, but it can still perform a (retrying) subscribe() call if the group exists and the topic isn’t in _bound_topics. That makes “passive” calls (e.g., from ack/reject/cancel paths) unexpectedly do network I/O and can also race between threads because _bound_topics is not protected by self.mutex. Consider either (1) skipping subscribe when passive=True, and (2) guarding the check/subscribe/add-to-_bound_topics section with the same lock to avoid duplicate subscriptions.

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

auvipy requested changes May 31, 2025

View reviewed changes

imzs and others added 2 commits May 31, 2025 18:49

Virtual transport of RocketMQ

95b102b

[pre-commit.ci] auto fixes from pre-commit.com hooks

65fb950

for more information, see https://pre-commit.ci

Nusnus force-pushed the main branch from 6f63103 to 65fb950 Compare May 31, 2025 15:49

Nusnus marked this pull request as draft May 31, 2025 15:59

fix unit test, BytesWarning

d06e41e

moling and others added 2 commits June 25, 2025 16:18

resolve dependency conflict: protobuf>=4.25.5, grpcio-tools>=1.67.0

8ad6076

Merge branch 'main' into resolve-conflict

4a52823

imzs marked this pull request as ready for review June 25, 2025 09:04

imzs requested a review from auvipy June 25, 2025 09:05

Merge branch 'main' into resolve-conflict

abe4acd

Merge branch 'main' into resolve-conflict

be45708

auvipy requested a review from Copilot November 9, 2025 16:41

auvipy added this to the 5.7.0 milestone Nov 9, 2025

Copilot AI reviewed Nov 9, 2025

View reviewed changes

fix typo, indentation, deprecated method

9027985

Change-Id: Ib0fcf35f367c3292ab415bcf46cfd4105c57248a

Merge branch 'main' into resolve-conflict

09bd4f8

Nusnus added 2 commits December 4, 2025 14:22

Merge branch 'main' into resolve-conflict

dcb38e2

Merge branch 'main' into resolve-conflict

840a8ee

auvipy requested changes Mar 1, 2026

View reviewed changes

Merge branch 'main' into resolve-conflict

9247cf7

auvipy requested a review from Copilot March 2, 2026 09:44

Copilot started reviewing on behalf of auvipy March 2, 2026 09:44 View session

Copilot AI reviewed Mar 2, 2026

View reviewed changes

Comment thread kombu/transport/rocketmq.py Outdated

Comment thread kombu/transport/rocketmq.py

Comment thread kombu/transport/rocketmq.py Outdated

Comment thread docs/reference/kombu.transport.rocketmq.rst Outdated

Comment thread kombu/transport/rocketmq.py

auvipy and others added 5 commits March 2, 2026 22:10

Update kombu/transport/rocketmq.py

76c2590

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Update kombu/transport/rocketmq.py

18eb1a8

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Update docs/reference/kombu.transport.rocketmq.rst

b086bdb

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Update kombu/transport/rocketmq.py

1f3b7a0

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

add rocketmq extra requirement

ee8b8ae

auvipy requested a review from Copilot March 7, 2026 11:51

Copilot started reviewing on behalf of auvipy March 7, 2026 11:51 View session

Copilot AI reviewed Mar 7, 2026

View reviewed changes

Comment thread kombu/transport/rocketmq.py

Comment thread kombu/transport/rocketmq.py

Comment thread kombu/transport/rocketmq.py Outdated

auvipy and others added 4 commits March 7, 2026 19:46

Update kombu/transport/rocketmq.py

e0748a5

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Update kombu/transport/rocketmq.py

3357694

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Merge branch 'main' into resolve-conflict

487c035

Merge branch 'main' into resolve-conflict

c5a1718

auvipy requested a review from Copilot March 25, 2026 14:04

Copilot started reviewing on behalf of auvipy March 25, 2026 14:04 View session

Copilot AI reviewed Mar 25, 2026

View reviewed changes

auvipy and others added 6 commits March 25, 2026 20:15

Update kombu/transport/rocketmq.py

dddae71

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Update docs/reference/kombu.transport.rocketmq.rst

acbf2b3

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Update kombu/transport/rocketmq.py

61f9dab

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Update kombu/transport/rocketmq.py

d5e37b8

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Update kombu/transport/rocketmq.py

4e6bfc4

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Update kombu/transport/rocketmq.py

cca60a4

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Uh oh!

Conversation

imzs commented May 30, 2025

Uh oh!

auvipy left a comment

Choose a reason for hiding this comment

Uh oh!

imzs commented Jun 1, 2025

Uh oh!

auvipy commented Jun 2, 2025

Uh oh!

imzs commented Jun 25, 2025

Uh oh!

auvipy commented Jun 25, 2025

Uh oh!

imzs commented Nov 6, 2025

Uh oh!

codecov bot commented Nov 9, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

imzs commented Nov 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

imzs commented Nov 24, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

imzs commented Jan 12, 2026

Uh oh!

auvipy left a comment

Choose a reason for hiding this comment

Uh oh!

imzs commented Mar 2, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI Mar 25, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI Mar 25, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 25, 2026

Choose a reason for hiding this comment

codecov bot commented Nov 9, 2025 •

edited

Loading

imzs commented Nov 11, 2025 •

edited

Loading

imzs commented Nov 24, 2025 •

edited

Loading