[RabbitMQ] Add async processing support - #4094
IllyaMoskvin wants to merge 1 commit into
Conversation
There was a problem hiding this comment.
Pull request overview
This PR extends Nuclio’s async processing mode to RabbitMQ triggers (Python runtime only), aligning RabbitMQ trigger behavior with the existing async/sync mode support patterns used by other triggers.
Changes:
- Allowlisted
rabbit-mqfor async trigger mode validation (Python runtime only). - Added async-mode execution path for RabbitMQ trigger message handling and worker allocation.
- Updated tests and documentation to reflect RabbitMQ async-mode support.
Reviewed changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| pkg/functionconfig/types.go | Adds rabbit-mq to the async-capable trigger kind allowlist. |
| pkg/processor/trigger/rabbitmq/factory.go | Selects worker allocator based on trigger mode (async vs sync/default). |
| pkg/processor/trigger/rabbitmq/trigger.go | Adds async message dispatch + graceful draining on stop; adds async-mode warning and helper. |
| pkg/processor/trigger/rabbitmq/rabbitmq_test.go | Adds unit tests for isAsyncMode() behavior. |
| pkg/platform/abstract/platform_test.go | Adds validation test coverage for rabbit-mq async mode on Python. |
| docs/tasks/async-mode.md | Updates supported trigger list to include RabbitMQ. |
| docs/reference/triggers/rabbitmq.md | Documents RabbitMQ async mode usage and caveats. |
Comments suppressed due to low confidence (2)
pkg/processor/trigger/rabbitmq/trigger.go:131
Stop()closesrmq.connectionErrorChaneven though it is also written/closed by the AMQP client viabrokerConn.NotifyClose(...). Closing a notification channel that another goroutine may send to can panic (send on closed channel). Also,Stop()closes the AMQP channel but does not closermq.brokerConn, which can leak connections. Prefer closing the connection (rmq.brokerConn.Close()), and let the AMQP library close/terminate the notify channel rather than closingconnectionErrorChandirectly.
// close broker
if err := rmq.brokerChannel.Close(); err != nil {
rmq.Logger.WarnWith("Failed to close broker channel", "err", err.Error())
}
close(rmq.connectionErrorChan)
return nil, nil
pkg/processor/trigger/rabbitmq/trigger.go:105
Start()initializeshandlerDonebeforecreateBrokerResources(), but ifcreateBrokerResources()returns an error the handler goroutine is never started andhandlerDoneis never closed. Any laterStop()will block forever on<-rmq.handlerDone. Consider either: (1) only creatinghandlerDoneafter broker resources succeed, (2) closinghandlerDoneon the error path, or (3) guardingStop()so it doesn't wait when the handler was never started.
rmq.stopChan = make(chan struct{})
rmq.handlerDone = make(chan struct{})
if err := rmq.createBrokerResources(); err != nil {
return errors.Wrap(err, "Failed to create broker resources")
}
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| case message := <-rmq.brokerInputMessagesChannel: | ||
| rmq.processMessage(&message) | ||
|
|
||
| // in async mode, dispatch each message in its own goroutine to allow concurrent processing | ||
| if rmq.isAsyncMode() { | ||
| rmq.inFlightMessages.Add(1) | ||
| go func() { | ||
| defer rmq.inFlightMessages.Done() | ||
| rmq.processMessage(&message) | ||
| }() | ||
| } else { | ||
| rmq.processMessage(&message) | ||
| } |
There was a problem hiding this comment.
handleBrokerMessages() receives from rmq.brokerInputMessagesChannel without checking the ok value. When the consume channel is closed (e.g., broker/channel closed or consumer canceled), this case will keep firing immediately with a zero-value delivery; in async mode it can also spawn unbounded goroutines. Read with message, ok := <-rmq.brokerInputMessagesChannel and exit (or reconnect) when !ok.
| if rmq.isAsyncMode() && rmq.configuration.PrefetchCount == 0 { | ||
| rmq.Logger.Warn("Async mode is enabled without a prefetchCount limit. " + | ||
| "This may cause unbounded goroutine creation under high throughput. " + | ||
| "Consider setting prefetchCount to limit concurrency") |
There was a problem hiding this comment.
The warning about missing prefetchCount implies it limits concurrency in async mode, but in the current implementation Qos(prefetchCount, ...) is only applied inside createTopics() when len(Topics) > 0 (it early-returns otherwise). That means a user can set prefetchCount with empty topics and still get unlimited deliveries without seeing this warning. Consider applying Qos whenever PrefetchCount != 0 (independent of topics), and/or adjusting the warning to reflect when prefetch will actually be enforced.
| if rmq.isAsyncMode() && rmq.configuration.PrefetchCount == 0 { | |
| rmq.Logger.Warn("Async mode is enabled without a prefetchCount limit. " + | |
| "This may cause unbounded goroutine creation under high throughput. " + | |
| "Consider setting prefetchCount to limit concurrency") | |
| if rmq.isAsyncMode() { | |
| if rmq.configuration.PrefetchCount == 0 { | |
| rmq.Logger.Warn("Async mode is enabled without a prefetchCount limit. " + | |
| "This may cause unbounded goroutine creation under high throughput. " + | |
| "Consider setting prefetchCount to limit concurrency") | |
| } else if len(rmq.configuration.Topics) == 0 { | |
| rmq.Logger.Warn("Async mode is enabled with prefetchCount configured, but no topics are set. " + | |
| "In the current configuration, prefetchCount may not be enforced, which can still lead to unbounded goroutine creation under high throughput") | |
| } |
TomerShor
left a comment
There was a problem hiding this comment.
Nice work, and thanks for the contribution!
I would like to see more of an integration test in pkg/processor/trigger/rabbitmq/test/rabbitmq_test.go, maybe something similar to the tests we have for async http handling, where we deploy a function and consume from the queue for real.
| rmq.Logger.Warn("Async mode is enabled without a prefetchCount limit. " + | ||
| "This may cause unbounded goroutine creation under high throughput. " + | ||
| "Consider setting prefetchCount to limit concurrency") |
There was a problem hiding this comment.
This code runs in the processor, meaning that this warning will only be printed in the function logs.
Consider moving this to a validation on the platform level, so you can see it on the build logs while building/deploying the function
rokatyy
left a comment
There was a problem hiding this comment.
Thanks for the contribution 🚀
Please add an integration test to make sure that it works e2e. Let me know if you need any guidance with that 🙏
| func (suite *TestSuite) TestIsAsyncModeReturnsFalseByDefault() { | ||
| suite.trigger.configuration = &Configuration{ | ||
| Configuration: trigger.Configuration{ | ||
| Trigger: &functionconfig.Trigger{}, | ||
| }, | ||
| } | ||
|
|
||
| suite.Require().False(suite.trigger.isAsyncMode()) | ||
| } | ||
|
|
||
| func (suite *TestSuite) TestIsAsyncModeReturnsFalseForSyncMode() { | ||
| suite.trigger.configuration = &Configuration{ | ||
| Configuration: trigger.Configuration{ | ||
| Trigger: &functionconfig.Trigger{ | ||
| Mode: functionconfig.SyncTriggerWorkMode, | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| suite.Require().False(suite.trigger.isAsyncMode()) | ||
| } | ||
|
|
||
| func (suite *TestSuite) TestIsAsyncModeReturnsTrueForAsyncMode() { | ||
| suite.trigger.configuration = &Configuration{ | ||
| Configuration: trigger.Configuration{ | ||
| Trigger: &functionconfig.Trigger{ | ||
| Mode: functionconfig.AsyncTriggerWorkMode, | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| suite.Require().True(suite.trigger.isAsyncMode()) | ||
| } | ||
|
|
There was a problem hiding this comment.
| func (suite *TestSuite) TestIsAsyncModeReturnsFalseByDefault() { | |
| suite.trigger.configuration = &Configuration{ | |
| Configuration: trigger.Configuration{ | |
| Trigger: &functionconfig.Trigger{}, | |
| }, | |
| } | |
| suite.Require().False(suite.trigger.isAsyncMode()) | |
| } | |
| func (suite *TestSuite) TestIsAsyncModeReturnsFalseForSyncMode() { | |
| suite.trigger.configuration = &Configuration{ | |
| Configuration: trigger.Configuration{ | |
| Trigger: &functionconfig.Trigger{ | |
| Mode: functionconfig.SyncTriggerWorkMode, | |
| }, | |
| }, | |
| } | |
| suite.Require().False(suite.trigger.isAsyncMode()) | |
| } | |
| func (suite *TestSuite) TestIsAsyncModeReturnsTrueForAsyncMode() { | |
| suite.trigger.configuration = &Configuration{ | |
| Configuration: trigger.Configuration{ | |
| Trigger: &functionconfig.Trigger{ | |
| Mode: functionconfig.AsyncTriggerWorkMode, | |
| }, | |
| }, | |
| } | |
| suite.Require().True(suite.trigger.isAsyncMode()) | |
| } | |
| func (suite *TestSuite) TestIsAsyncMode() { | |
| testCases := []struct { | |
| name string | |
| mode functionconfig.TriggerWorkMode | |
| expected bool | |
| }{ | |
| { | |
| name: "default mode (empty) returns false", | |
| mode: "", | |
| expected: false, | |
| }, | |
| { | |
| name: "sync mode returns false", | |
| mode: functionconfig.SyncTriggerWorkMode, | |
| expected: false, | |
| }, | |
| { | |
| name: "async mode returns true", | |
| mode: functionconfig.AsyncTriggerWorkMode, | |
| expected: true, | |
| }, | |
| } | |
| for _, tc := range testCases { | |
| suite.Run(tc.name, func() { | |
| suite.trigger.configuration = &Configuration{ | |
| Configuration: trigger.Configuration{ | |
| Trigger: &functionconfig.Trigger{ | |
| Mode: tc.mode, | |
| }, | |
| }, | |
| } | |
| result := suite.trigger.isAsyncMode() | |
| suite.Require().Equal(tc.expected, result) | |
| }) | |
| } | |
| } |
| if rmq.isAsyncMode() { | ||
| rmq.inFlightMessages.Add(1) | ||
| go func() { | ||
| defer rmq.inFlightMessages.Done() | ||
| rmq.processMessage(&message) | ||
| }() | ||
| } else { |
There was a problem hiding this comment.
How about a semaphore instead? ( we cannot process more than max connections at the same time anyways)
|
|
||
| // now wait for any in-flight async message goroutines to complete | ||
| // before closing the broker channel, so they can properly ack/nack their messages | ||
| rmq.inFlightMessages.Wait() |
There was a problem hiding this comment.
we should also give it some time in case of reconnect (see reconnect())
|
@TomerShor @rokatyy Thank you both for the review! I'm a bit slammed at work this week, but I think I can do another pass on this towards the end of this week, or over this upcoming weekend. |
📝 Description
Add support for asynchronous processing mode on RabbitMQ triggers (Python runtime only). Previously, async mode was restricted to HTTP triggers. This change enables RabbitMQ triggers to process messages concurrently when configured with
mode: async, while preserving the existing serial processing behavior for sync (default) mode.🛠️ Changes Made
pkg/functionconfig/types.go: Added"rabbit-mq"to thetriggerKindsSupportAsyncallowlist, enabling async mode at the platform validation layer.pkg/processor/trigger/rabbitmq/factory.go: Worker allocator creation now switches on trigger mode:NonBlockingWorkerAllocatorfor async,FixedPoolWorkerAllocatorfor sync/default. This mirrors the existing HTTP factory pattern.pkg/processor/trigger/rabbitmq/trigger.go:handleBrokerMessages(), async mode dispatchesprocessMessagein its own goroutine for concurrent processing. Sync mode continues to process messages serially.sync.WaitGroup(inFlightMessages) to track in-flight async goroutines and drain them gracefully inStop()before closing the broker channel.Initialize()when async mode is used without aprefetchCountlimit.isAsyncMode()helper.pkg/processor/trigger/rabbitmq/rabbitmq_test.go: Three new unit tests verifyingisAsyncMode()returns correct values for default/empty, explicit sync, and async modes.pkg/platform/abstract/platform_test.go: New test case confirmingrabbit-mqwithAsyncTriggerWorkModepasses platform validation.docs/tasks/async-mode.md: Updated supported triggers list to include RabbitMQ alongside HTTP.docs/reference/triggers/rabbitmq.md: Added "Asynchronous Mode" section with configuration details, ordering caveats, and an example.✅ Checklist
🧪 Testing
go test -tags test_unit):pkg/processor/trigger/rabbitmq/...– 10/10 passing (6 existing + 3 newisAsyncModetests + 1 existing config test)pkg/platform/abstract–TestValidateProcessingMode– 9/9 passing (8 existing + 1 newrabbit-mq async trigger with valid config)if rmq.isAsyncMode()branch only activates when the trigger is explicitly configured withmode: async.🔗 References
🚨 Breaking Changes?
🔍️ Additional Notes
runtimesSupportAsyncallowlist (Python only) enforces this, so no changes needed.prefetchCountis 0 (unlimited). In async mode, each message spawns a goroutine, so running without a prefetch limit can cause unbounded goroutine creation. The trigger logs a warning at startup if async mode is enabled without aprefetchCount. Users should set it explicitly.Stop()first waits for the message handler loop to exit via ahandlerDonechannel (preventing new goroutines from being spawned), then waits on async.WaitGroupto drain in-flight async goroutines, and only then closes the broker channel. This ensures all messages are properly ACKed/NACKed before teardown.processMessagegoroutine independently blocks until the worker returns, then ACKs/NACKs. Theamqp091-golibrary supports concurrent ACK/NACK on the same channel.