feat(connector): Add topic to mqtt additional columns #19017

Boudewijn26 · 2024-10-18T14:12:51Z

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

Make the MQTT topic available as an additional column
Revert the change made in fix: prevent mqtt enumerator in dead loop if disconnected #18961 to restore MQTT functionality

Checklist

I have written necessary rustdoc comments
I have added necessary unit tests and integration tests
I have added test labels as necessary. See details.
I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
All checks passed in ./risedev check (or alias, ./risedev c)
My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

It's now possible to include the topic on which MQTT messages have been received by using INCLUDE topic in your SELECT

tabVersion · 2024-10-21T07:41:54Z

src/connector/src/parser/additional_columns.rs

@@ -87,6 +87,10 @@ pub static COMPATIBLE_ADDITIONAL_COLUMNS: LazyLock<HashMap<&'static str, HashSet
                    "collection_name",
                ]),
            ),
+            (
+                MQTT_CONNECTOR,
+                HashSet::from(["topic", "offset", "partition"]),


IIUC, topic and partition are the same thing in the MQTT connector?

Yes, I followed the pattern established by the file connectors, which have "file" the same as the "partition" and read the split_id as the file

https://github.com/risingwavelabs/risingwave/blob/main/src/connector/src/parser/mod.rs#L484-L491

Let's use partition instead. For other connectors, split_id is the partition number, no need for another additional column.

tabVersion · 2024-10-21T07:53:34Z

src/connector/src/parser/mod.rs

+                    return Ok(A::output_for(
+                        self.row_meta
+                            .as_ref()
+                            .map(|ele| ScalarRefImpl::Utf8(ele.split_id)),


What if subscribing to foo.*? I think the topic info should come from each message, you can take the kafka source impl as example.

The topic is saved in split_id, just like the "file" additional column is also stored in the split_id:
https://github.com/risingwavelabs/risingwave/blob/main/src/connector/src/source/mqtt/source/reader.rs#L96
https://github.com/risingwavelabs/risingwave/blob/main/src/connector/src/source/mqtt/source/message.rs#L41-L47
https://github.com/risingwavelabs/risingwave/blob/main/src/connector/src/source/mqtt/source/reader.rs#L97
https://github.com/risingwavelabs/risingwave/blob/main/src/connector/src/source/mqtt/source/message.rs#L29-L37

tabVersion · 2024-10-21T08:15:35Z

src/connector/src/source/mqtt/enumerator/mod.rs

+            loop {
+                if self.connected.load(std::sync::atomic::Ordering::Relaxed) {
+                    break;
+                }
+
+                if start.elapsed().as_secs() > 10 {
+                    bail!("Failed to connect to mqtt broker");
+                }
+
+                tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+            }


regarding the comments #18961 (comment), I still have some concern about the wait time. Would you be able to move the logic in the tokio thread into this func?

I don't fully understand exactly what you're suggesting, but I'd say it's one of:

Move the logic inside the tokio thread to the list_splits, spawn the thread in new

During the call to list_splits spawn a thread and then stop that thread when the topics have come in

During the call to list_splits spawn a new thread to listen to the incoming mqtt topics

During the call to list_splits spawn a new thread if one isn't already running to listen to the incoming mqtt topics

Option 3 would spawn a new thread on each enumerator tick, which doesn't seem desirable to me
Option 4 is functionally identical to what we have just with added state to track the existence of the thread; I don't really see what would be gained
Option 2 would risk missing topics that come in after list_splits returns and come in before the next tick
Option 1 would require some inter thread communication in order to poll the mqtt eventloop inside the spawned thread whilst having the logic in the list_splits. That seems to make it more complicated than desired

Please indicate if I've misunderstood

Option 2 would risk missing topics that come in after list_splits returns and come in before the next tick

Oh, I got it. Mqtt does not have persisted metadata so RisingWave needs a thread running in the background. The fix makes sense.

tabVersion

Please reuse partition for topic in MQTT connector. The concept is hard to align with other connectors.

LGTM, thanks for your contribution.

risingwavelabs/risingwave#19017

Boudewijn26 · 2024-10-22T11:08:39Z

@tabVersion With ae9f988 (#19017) I've modified this so that it uses the partition instead of a separate topic column.

Meanwhile foundation-zero/risingwave-docs@8d2b58f is ready for a pull request once this is merged

yufansong

LGTM

tabVersion · 2024-10-29T06:14:47Z

Let's merge

Boudewijn26 force-pushed the feature/mqtt-topic-column branch 2 times, most recently from a85a7b8 to a0aed41 Compare October 21, 2024 07:08

feat(connector): Add topic to mqtt additional columns

31ab0a5

Boudewijn26 force-pushed the feature/mqtt-topic-column branch from a0aed41 to 31ab0a5 Compare October 21, 2024 07:12

tabVersion requested review from tabVersion and yufansong October 21, 2024 07:14

graphite-app bot requested a review from a team October 21, 2024 07:54

tabVersion reviewed Oct 21, 2024

View reviewed changes

tabVersion approved these changes Oct 22, 2024

View reviewed changes

Boudewijn26 added a commit to foundation-zero/risingwave-docs that referenced this pull request Oct 22, 2024

Add additional column for MQTT topic docs

8d2b58f

risingwavelabs/risingwave#19017

Change MQTT topic additional column to partition

ae9f988

yufansong approved these changes Oct 22, 2024

View reviewed changes

Reset mqtt broker in e2e test to mqtt-server instead of localhost

d3a6caf

Boudewijn26 force-pushed the feature/mqtt-topic-column branch from 7604448 to d3a6caf Compare October 23, 2024 10:42

tabVersion mentioned this pull request Oct 29, 2024

fix(ci): fix mqtt ci #19079

Closed

9 tasks

xxhZs added the ci/run-e2e-mqtt-sink-tests label Oct 29, 2024

xxhZs approved these changes Oct 29, 2024

View reviewed changes

tabVersion added this pull request to the merge queue Oct 29, 2024

Merged via the queue into risingwavelabs:main with commit 661939a Oct 29, 2024
34 of 35 checks passed

Boudewijn26 deleted the feature/mqtt-topic-column branch October 31, 2024 14:03

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(connector): Add topic to mqtt additional columns #19017

feat(connector): Add topic to mqtt additional columns #19017

Boudewijn26 commented Oct 18, 2024 •

edited

Loading

tabVersion Oct 21, 2024

Boudewijn26 Oct 21, 2024

tabVersion Oct 22, 2024

tabVersion Oct 21, 2024

Boudewijn26 Oct 21, 2024

tabVersion Oct 21, 2024

Boudewijn26 Oct 21, 2024

tabVersion Oct 22, 2024

tabVersion left a comment

Boudewijn26 commented Oct 22, 2024

yufansong left a comment

tabVersion commented Oct 29, 2024

feat(connector): Add topic to mqtt additional columns #19017

feat(connector): Add topic to mqtt additional columns #19017

Conversation

Boudewijn26 commented Oct 18, 2024 • edited Loading

What's changed and what's your intention?

Checklist

Documentation

Release note

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

tabVersion left a comment

Choose a reason for hiding this comment

Boudewijn26 commented Oct 22, 2024

yufansong left a comment

Choose a reason for hiding this comment

tabVersion commented Oct 29, 2024

Boudewijn26 commented Oct 18, 2024 •

edited

Loading