Feature: add support for ingesting from rabbitmq super streams

[jamiechapmanbrn]

This PR aims to add support for ingesting logs from a rabbitmq superstream, as a peer to Kafka or Kinesis. Since this supports scalable exactly once delivery, this provides a good gasket between RabbitMQ and Druid.

Currently, this is a WIP for a few reasons:

1. No support for username/password and some other features, I welcome suggestions here as I'm not exactly sure how those work yet and could use some guidance on how that plumbing works

2. The algorithm is pretty poor here, simply making a new rabbit consumer for each poll, it should be doing something similar to Kinesis where it fetches in the background, occasionally backing off if it gets too far ahead.

3. Testing is extremely bare bones, primarily because the design of the record supplier could change here, and design layout could change test layout here.

4. Documentation on how exactly to use it is missing. I've been testing it with the below supervisor spec, the primary configuration mechanism is the 'uri' which is used to configure how it connects to rabbitmq for reading metadata and for getting messages

The meat of the change is in RabbitStreamRecordSupplier, which has the components that interface with rabbitmq.
It uses two interfaces, one being the low level 'Client' interface to read the partitions from a super stream so that druid can distribute work, and the other being the 'Consumer' interface, the conventional interface which creates a thread that listens for messages for you, and calls back on the handler which puts the messages in queue.

The current interface simply manages a map of ConsumerBuilders as partitions are assigned, and then when it is polled for messages it will create consumers and wait for the timeout. Once the timeout has elapsed, it will close all the consumers, then collect any messages from the queue and return them. This should ensure no messages are lost, but probably isn't very efficient as it could be reconnecting more than necessary.

Release note

New: You can now ingest logs from RabbitMQ via super streams

This feature gives the ability to read directly from rabbitmq using the new super-streams feature. As super streams allows exactly-once delivery with full support for partitioning, it is now compatible with Druid's modern ingestion algorithm, without the downsides that the prior RabbitMQ firehose.

Note that this uses the RabbitMQ streams feature, and not a conventional exchange. You will need to make sure that your messages are in a super stream before consumption. For more information, see https://www.rabbitmq.com/streams.html.

In order to configure, create a new index task using the 'rabbit' type, and configure it's stream and URI to connect to the rabbitmq host.

{
  "type": "rabbit",
  "dataSchema": {
    "dataSource": <your super-stream>,
    }
  },
  "ioConfig": {
    "type": "rabbit",
    "stream": "api-audit",
    "uri": "rabbitmq-stream://localhost:5552",
    "taskCount": 1,
    "replicas": 1,
    "taskDuration": "PT1H"
  },
  "tuningConfig": {
    "type": "rabbit",
    "resetOffsetAutomatically": true
  }
}

---

Key changed/added classes in this PR

The following classes have been added to druid in this PR:

• RabbitStreamSamplerSpec
• RabbitStreamRecordSupplier
• RabbitStreamIndexTaskTuningConfig
• RabbitStreamIndexTaskModule
• RabbitStreamIndexTaskIOConfig
• RabbitStreamIndexTaskClientFactory
• RabbitStreamIndexTask
• RabbitStreamDataSourceMetadata
• IncrementalPublishingRabbitStreamIndexTaskRunner
• RabbitSupervisorTuningConfig
• RabbitStreamSupervisorSpec
• RabbitStreamSupervisorReportPayload
• RabbitStreamSupervisorIOConfig
• RabbitStreamSupervisorIngestionSpec
• RabbitStreamSupervisor

---

This PR has:

• been self-reviewed.
  • using the concurrency checklist (Remove this item if the PR doesn't have any relation to concurrency.)
• added documentation for new or modified features or behaviors.
• a release note entry in the PR description.
• [] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
• added or updated version, license, or notice information in licenses.yaml
• added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
• added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
• added integration tests.
• been tested in a test Druid cluster.

[abhishekagarwal87]

@jamiechapmanbrn - let us know if you need any help from druid dev community in moving this PR forward.

[jamiechapmanbrn]

I've marked this as ready for review. We're testing it currently in our cluster and it seems to be working fine. I've added documentation and extended the tests significantly, I'm not 100% sure whether it has enough coverage to pass the CI yet or not, but I'm running out of steam to work on those. They are relatively tricky for me since I don't work much with Java and I'm not as familiar with the supervisor structures/classes.

[github-advanced-security]

CodeQL found more than 10 potential problems in the proposed changes. Check the Files changed tab for more details.

[abhishekagarwal87]

have you done any scale testing on this yet? I am wondering what kind of scale you have run it with so far.

[jamiechapmanbrn]

Currently, our org is running with it with a relatively low scale, somewhere on the order of a few events per minute. What would you consider reasonable for scale testing?

[AmatyaAvadhanula]

@jamiechapmanbrn, I'll complete my review for this PR soon.
Could you please take a look at the Integration tests that are failing as well?

[jamiechapmanbrn]

Okay, I rebased on the latest master and updated the pom.xml file to reference the latest. I'll have a look at any test failures that come when they're available.

[github-advanced-security]

CodeQL found more than 20 potential problems in the proposed changes. Check the Files changed tab for more details.

[AmatyaAvadhanula]

Thank you @jamiechapmanbrn. Overall approach LGTM!
Have a few minor comments that I'll post soon.

[AmatyaAvadhanula]

Could you please take a pass over the RabbitStreamRecordSupplier and RabbitStreamSupervisor classes to clean up minor things such as unneeded exceptions, unused arguments in newly added methods, and see if any public methods exist that can be made private?

[AmatyaAvadhanula]

LGTM!
+1 after the conflicts and build failures are fixed.

[jamiechapmanbrn]

That's great to hear!

I've rebased once again, and added additional test coverage. It should pass now. I'll take another look once the CI has done its magic.

[jamiechapmanbrn]

I think I've fixed the pipeline issues.
I see a bunch of dependency pull problems for druid-iceberg-extensions that don't have any relation to this pr.

Error: Exception in thread "main" java.lang.RuntimeException: java.lang.RuntimeException: org.eclipse.aether.resolution.DependencyResolutionException: Could not find artifact org.apache.druid.extensions:druid-iceberg-extensions:jar:30.0.0-SNAPSHOT in  (https://repo1.maven.org/maven2/)
	at org.apache.druid.cli.PullDependencies.run(PullDependencies.java:333)
	at org.apache.druid.cli.Main.main(Main.java:112)
Caused by: java.lang.RuntimeException: org.eclipse.aether.resolution.DependencyResolutionException: Could not find artifact org.apache.druid.extensions:druid-iceberg-extensions:jar:30.0.0-SNAPSHOT in  (https://repo1.maven.org/maven2/)
	at org.apache.druid.cli.PullDependencies.downloadExtension(PullDependencies.java:418)
	at org.apache.druid.cli.PullDependencies.downloadExtension(PullDependencies.java:363)
	at org.apache.druid.cli.PullDependencies.run(PullDependencies.java:309)
	... 1 more

[AmatyaAvadhanula]

@jamiechapmanbrn Thank you!

[jamiechapmanbrn]

That's great! Thanks for all the help!