feat(event): opt-in dual-sink for native queue and event plugin#6878
feat(event): opt-in dual-sink for native queue and event plugin#6878perfch wants to merge 1 commit into
Conversation
20c3c5b to
0e13b2f
Compare
There was a problem hiding this comment.
Pull request overview
This PR introduces an explicit, backward-compatible “dual-sink” event delivery mode where the native ZMQ queue and the event plugin (e.g., MongoDB) can run simultaneously, without inferring plugin activation from a stale path.
Changes:
- Adds explicit plugin activation resolution (
runPluginWithNativeQueue) and plugin load failure policy (pluginLoadFailurePolicy=fail|ignore). - Updates
EventPluginLoaderto support dual-sink routing, shared per-trigger serialization, plugin fault isolation, and back-pressure semantics in dual-sink mode. - Updates config templates and adds/extends unit tests for upgrade regression, routing, isolation, and load-failure behavior.
Reviewed changes
Copilot reviewed 8 out of 8 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| framework/src/test/java/org/tron/core/event/EventPluginLoaderTest.java | Adds regression/routing/isolation/load-failure tests for explicit dual-sink/plugin activation behavior. |
| framework/src/main/resources/config.conf | Documents opt-in dual-sink flag and plugin load failure policy in the sample config. |
| framework/src/main/java/org/tron/core/services/event/HistoryEventService.java | Ensures plugin back-pressure is respected before native queue pacing (incl. dual-sink mode). |
| framework/src/main/java/org/tron/core/config/args/Args.java | Resolves plugin activation explicitly via EventPluginConfig.resolveUseEventPlugin(...) and wires failure policy. |
| framework/src/main/java/org/tron/common/logsfilter/EventPluginLoader.java | Implements dual-sink routing, plugin activation flag, failure policy behavior, and shared serialization. |
| common/src/main/resources/reference.conf | Adds new config keys with defaults to the reference configuration. |
| common/src/main/java/org/tron/core/config/args/EventConfig.java | Adds bean fields for runPluginWithNativeQueue and pluginLoadFailurePolicy. |
| common/src/main/java/org/tron/common/logsfilter/EventPluginConfig.java | Adds useEventPlugin, pluginLoadFailurePolicy, and centralized activation resolution helper. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| private void postToPlugin(Consumer<IPluginEventListener> handler) { | ||
| if (Objects.isNull(eventListeners)) { | ||
| return; | ||
| } | ||
| try { | ||
| eventListeners.forEach(handler); | ||
| } catch (Exception e) { | ||
| logger.error("event plugin failed to handle trigger", e); | ||
| } | ||
| } |
There was a problem hiding this comment.
Addressed in 3c03c6d. postToPlugin now isolates per listener (each handler.accept(listener) in its own try/catch), so one failing plugin no longer blocks delivery to the others or the event thread. I intentionally catch Exception rather than Throwable: OutOfMemoryError/linkage Errors should propagate and fail the process, not be silently swallowed. Added a two-listener test (testPluginFailureIsIsolated) asserting the healthy listener still receives the trigger.
| useNativeQueue = config.isUseNativeQueue(); | ||
| useEventPlugin = config.isUseEventPlugin(); | ||
| if (StringUtils.hasText(config.getPluginLoadFailurePolicy())) { | ||
| pluginLoadFailurePolicy = config.getPluginLoadFailurePolicy().trim(); | ||
| } |
There was a problem hiding this comment.
Addressed in 3c03c6d. Added normalizeFailurePolicy(): only an explicit "ignore" enables ignore behavior; any unknown/typo value normalizes to the safe "fail" default (with a warning log). The policy is also re-resolved on every start(), so no policy state leaks across runs. Covered by testLoadFailurePolicyIgnoreDisablesPlugin, which asserts an unknown value normalizes to "fail".
| if (useNativeQueue) { | ||
| return launchNativeQueue(config); | ||
| } |
There was a problem hiding this comment.
Good catch, addressed in 3c03c6d. When launchNativeQueue(config) fails after the plugin has already started, start() now calls stopPlugin() before returning false, so a failed startup no longer leaks the plugin threads/resources.
| // Plugin connection settings are always copied when a path is configured, so the | ||
| // plugin can run either on its own (native queue off) or alongside the native queue. | ||
| boolean hasPluginPath = StringUtils.isNotEmpty(ec.getPath()); | ||
| if (hasPluginPath) { | ||
| epc.setPluginPath(ec.getPath().trim()); | ||
| } | ||
| if (StringUtils.isNotEmpty(ec.getServer())) { | ||
| epc.setServerAddress(ec.getServer().trim()); | ||
| } | ||
| if (StringUtils.isNotEmpty(ec.getDbconfig())) { | ||
| epc.setDbConfig(ec.getDbconfig().trim()); | ||
| } |
There was a problem hiding this comment.
Addressed in 3c03c6d. server/dbconfig are now copied only under hasPluginPath, matching the comment. Without a plugin path the plugin cannot be activated, so these connection settings are left unset rather than lingering on EventPluginConfig.
| public class EventPluginLoaderTest { | ||
|
|
||
| private static IPluginEventListener mockListener(EventPluginLoader loader) { | ||
| IPluginEventListener listener = mock(IPluginEventListener.class); | ||
| List<IPluginEventListener> list = new ArrayList<>(); | ||
| list.add(listener); | ||
| ReflectUtils.setFieldValue(loader, "eventListeners", list); | ||
| return listener; | ||
| } |
There was a problem hiding this comment.
Addressed in 3c03c6d. Added an @After resetLoader() that resets the singletons useNativeQueue/useEventPlugin/eventListeners/pluginLoadFailurePolicy after each test, preventing state from leaking into other test classes that use EventPluginLoader.getInstance().
Allow the native ZMQ queue and the event plugin to run at the same time,
with activation made explicit rather than inferred from a path so that
existing native-queue deployments keep their current behavior after an
upgrade.
- Plugin activation is resolved explicitly in Args via
EventPluginConfig.resolveUseEventPlugin(): with the native queue on it
only joins when the operator opts in (runPluginWithNativeQueue = true),
so a stale "path" in an upgraded config can never silently load a plugin.
- Add pluginLoadFailurePolicy ("fail" | "ignore") to define behavior when
the plugin cannot load; default "fail" preserves data integrity.
- Isolate plugin trigger posting in try/catch so a plugin fault cannot tear
down the node event thread; serialize each trigger once and share it with
both sinks so the default path keeps its current cost.
- Apply plugin back-pressure (isBusy) whenever the plugin is active,
including dual-sink mode, in EventPluginLoader and HistoryEventService.
- Tests: upgrade regression, dual-sink routing, plugin-failure isolation,
load-failure policy, and updated back-pressure semantics.
0e13b2f to
3c03c6d
Compare
What does this PR do?
Lets the native ZMQ queue and the event plugin (e.g. MongoDB) run at the same time, so a node can have both low-latency ZMQ delivery and durable plugin persistence from a single configuration. Activation is made explicit instead of being inferred from the plugin
path.event.subscribe.runPluginWithNativeQueue(defaultfalse). With the native queue enabled, the plugin only joins when this istrue. When the native queue is disabled, behavior is unchanged (the plugin remains the only sink).event.subscribe.pluginLoadFailurePolicy(fail|ignore, defaultfail) to define what happens when the plugin cannot load.Argsresolves activation viaEventPluginConfig.resolveUseEventPlugin(...); the loader no longer infers activation from a raw path.isBusy) is honored whenever the plugin is active, including dual-sink mode, both at runtime and during historical event sync (HistoryEventService).This builds on the idea from #6852 and is an alternative that addresses the backward-compatibility and isolation concerns raised there. Related: #6862.
Why are these changes required?
Previously the native queue and the plugin were mutually exclusive: with
useNativeQueue = truethe plugin config was ignored entirely, forcing operators to choose between low latency and durable storage.The main risk with enabling both is an upgrade regression: a node that still carries a stale
pathin its config must not suddenly start loading a plugin after an upgrade. Making activation an explicit opt-in keeps every existing native-queue deployment on its current behavior by default, while giving operators who want dual-sink a clear switch. The failure policy and the try/catch isolation keep a plugin problem from taking the node down.Review concern → how it is addressed:
pathin an upgraded native-queue config would silently activate the plugin (backward-compat regression)runPluginWithNativeQueueopt-in (defaultfalse); a path alone never activates the plugin when the native queue is onpluginLoadFailurePolicy=fail(default, abort startup) orignore(log and continue native-queue-only)This PR has been tested by:
EventPluginLoaderTest:testResolveUseEventPlugin(upgrade regression),testIsBusy(back-pressure incl. dual-sink),testStartRequiresASink,testPostRoutesToPluginOnlyWhenActive,testPluginFailureIsIsolated,testLoadFailurePolicyIgnoreDisablesPluginArgsTestandParameterTestpass (config binding of the two new keys is backward compatible)checkstyleMain/checkstyleTestpassFollow up
Documentation for combined mode is covered inline in
config.conf. A separate docs page update can follow if desired.Extra details
Default behavior is unchanged: with
runPluginWithNativeQueue = false(the default), a native-queue node behaves exactly as before, even if apathis still present in its config.