Skip to content

Commit 99844ee

Browse files
kjnilssonkjnilsson
committed
Make coordination Ra system parameters configurable via rabbitmq.conf
Expose all coordination Ra system parameters through new coordination.* configuration options, following the pattern established in PR #15962 for quorum queues. This allows operators to fine-tune Raft log and WAL behavior specifically for the coordination system without affecting other Ra-based systems (quorum queues, Khepri). New coordination.* configuration options: - coordination.wal_compute_checksums (default: true) - coordination.segment_compute_checksums (default: true) - coordination.wal_max_size_bytes (default: 64 MB) - coordination.wal_max_entries (default: from Ra's default_config) - coordination.wal_max_batch_size (default: from Ra's default_config) - coordination.segment_max_size_bytes (default: from Ra's default_config) - coordination.segment_max_entries (default: from Ra's default_config) - coordination.max_append_entries_rpc_batch_size (default: from Ra's default_config) - coordination.compress_mem_tables (default: true) - coordination.snapshot_chunk_size (default: from Ra's default_config) Each coordination.* setting has a fallback to its default value, ensuring backward compatibility. The wal_max_size_bytes parameter maintains the existing hardcoded default of 64 MB. Implementation details: 1. Added macros for coordination defaults in rabbit_ra_systems.erl 2. Refactored get_config(coordination) to read from application environment variables with sensible defaults, matching the quorum_queues pattern 3. Added Cuttlefish schema mappings for all new configuration options 4. Added comprehensive documentation in rabbitmq.conf.example 5. Added schema validation tests in rabbit.snippets 6. Added dedicated test suite (rabbit_ra_systems_SUITE) to validate defaults All configuration is explicitly passed to the coordination Ra system during initialization, making the data flow transparent and maintainable. Backward compatibility: Existing deployments continue to work unchanged, with coordination.wal_max_size_bytes defaulting to 64 MB.
1 parent 65a5433 commit 99844ee

6 files changed

Lines changed: 318 additions & 2 deletions

File tree

‎deps/rabbit/Makefile‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -251,7 +251,7 @@ endef
251251
PARALLEL_CT_SET_1_A = unit_rabbit_ssl unit_cluster_formation_locking_mocks unit_cluster_formation_sort_nodes unit_collections unit_config_value_encryption unit_connection_tracking
252252
PARALLEL_CT_SET_1_B = amqp_address amqp_auth amqp_filter_prop amqp_filter_sql amqp_filter_sql_unit amqp_dotnet signal_handling single_active_consumer unit_access_control_authn_authz_context_propagation unit_access_control_credential_validation unit_amqp091_content_framing unit_amqp091_server_properties unit_app_management
253253
PARALLEL_CT_SET_1_C = amqp_proxy_protocol amqpl_consumer_ack bindings rabbit_db_maintenance rabbit_db_msup rabbit_db_policy rabbit_db_queue rabbit_db_topic_exchange cluster_limit cluster_minority default_queue_type_prop term_to_binary_compat_prop topic_permission unicode unit_access_control
254-
PARALLEL_CT_SET_1_D = amqqueue_backward_compatibility channel_interceptor channel_operation_timeout classic_queue_prop config_schema peer_discovery_dns peer_discovery_tmp_hidden_node per_node_limit per_user_connection_channel_limit
254+
PARALLEL_CT_SET_1_D = amqqueue_backward_compatibility channel_interceptor channel_operation_timeout classic_queue_prop config_schema peer_discovery_dns peer_discovery_tmp_hidden_node per_node_limit per_user_connection_channel_limit rabbit_ra_systems
255255

256256
PARALLEL_CT_SET_2_A = cluster confirms_rejects consumer_timeout rabbit_access_control rabbit_confirms rabbit_core_metrics_gc rabbit_cuttlefish rabbit_db_binding rabbit_db_exchange
257257
PARALLEL_CT_SET_2_B = crashing_queues deprecated_features direct_exchange_routing_v2 disconnect_detected_during_alarm exchanges unit_gen_server2

‎deps/rabbit/docs/rabbitmq.conf.example‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -527,6 +527,54 @@
527527
##
528528
# quorum_queue.snapshot_chunk_size = 1000000
529529

530+
##
531+
## Coordination System Ra Parameters
532+
##
533+
## Relevant doc guides:
534+
##
535+
## * https://www.rabbitmq.com/docs/clustering
536+
##
537+
538+
## Coordination system WAL compute checksums flag.
539+
##
540+
# coordination.wal_compute_checksums = true
541+
542+
## Coordination system segment compute checksums flag.
543+
##
544+
# coordination.segment_compute_checksums = true
545+
546+
## Maximum size in bytes of each WAL file segment for the coordination system.
547+
##
548+
# coordination.wal_max_size_bytes = 64000000
549+
550+
## Maximum number of entries allowed in the coordination system WAL.
551+
##
552+
# coordination.wal_max_entries = 500000
553+
554+
## Maximum number of entries allowed in a single WAL write batch for the coordination system.
555+
##
556+
# coordination.wal_max_batch_size = 4096
557+
558+
## Maximum size in bytes of each log segment for the coordination system.
559+
##
560+
# coordination.segment_max_size_bytes = 64000000
561+
562+
## Maximum number of entries in a log segment for the coordination system.
563+
##
564+
# coordination.segment_max_entries = 4096
565+
566+
## Maximum number of append entries RPC batch size for the coordination system.
567+
##
568+
# coordination.max_append_entries_rpc_batch_size = 16
569+
570+
## Whether to compress memory tables in the coordination system.
571+
##
572+
# coordination.compress_mem_tables = true
573+
574+
## Size of snapshot chunks in bytes for the coordination system.
575+
##
576+
# coordination.snapshot_chunk_size = 1000000
577+
530578
## Changes classic queue storage implementation version.
531579
## As of 4.0.0, version 2 is the default and this is a forward compatibility setting,
532580
## that is, it will be useful when a new version is developed.

‎deps/rabbit/priv/schema/rabbit.schema‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2824,6 +2824,58 @@ end}.
28242824
]}.
28252825

28262826

2827+
%%
2828+
%% Coordination System Ra Parameters
2829+
%%
2830+
2831+
{mapping, "coordination.wal_compute_checksums", "rabbit.coordination_wal_compute_checksums", [
2832+
{datatype, {enum, [true, false]}}
2833+
]}.
2834+
2835+
{mapping, "coordination.segment_compute_checksums", "rabbit.coordination_segment_compute_checksums", [
2836+
{datatype, {enum, [true, false]}}
2837+
]}.
2838+
2839+
{mapping, "coordination.wal_max_size_bytes", "rabbit.coordination_wal_max_size_bytes", [
2840+
{datatype, integer},
2841+
{validators, ["non_zero_positive_integer"]}
2842+
]}.
2843+
2844+
{mapping, "coordination.wal_max_entries", "rabbit.coordination_wal_max_entries", [
2845+
{datatype, integer},
2846+
{validators, ["non_zero_positive_integer"]}
2847+
]}.
2848+
2849+
{mapping, "coordination.wal_max_batch_size", "rabbit.coordination_wal_max_batch_size", [
2850+
{datatype, integer},
2851+
{validators, ["non_zero_positive_integer"]}
2852+
]}.
2853+
2854+
{mapping, "coordination.segment_max_size_bytes", "rabbit.coordination_segment_max_size_bytes", [
2855+
{datatype, integer},
2856+
{validators, ["non_zero_positive_integer"]}
2857+
]}.
2858+
2859+
{mapping, "coordination.segment_max_entries", "rabbit.coordination_segment_max_entries", [
2860+
{datatype, integer},
2861+
{validators, ["non_zero_positive_integer", "positive_16_bit_unsigned_integer"]}
2862+
]}.
2863+
2864+
{mapping, "coordination.max_append_entries_rpc_batch_size", "rabbit.coordination_max_append_entries_rpc_batch_size", [
2865+
{datatype, integer},
2866+
{validators, ["non_zero_positive_integer"]}
2867+
]}.
2868+
2869+
{mapping, "coordination.compress_mem_tables", "rabbit.coordination_compress_mem_tables", [
2870+
{datatype, {enum, [true, false]}}
2871+
]}.
2872+
2873+
{mapping, "coordination.snapshot_chunk_size", "rabbit.coordination_snapshot_chunk_size", [
2874+
{datatype, integer},
2875+
{validators, ["non_zero_positive_integer"]}
2876+
]}.
2877+
2878+
28272879
%%
28282880
%% Runtime parameters
28292881
%%

‎deps/rabbit/src/rabbit_ra_systems.erl‎

Lines changed: 35 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,10 @@
2929
%% the default min bin vheap value in OTP 26
3030
-define(MIN_BIN_VHEAP_SIZE_DEFAULT, 46422).
3131
-define(MIN_BIN_VHEAP_SIZE_MULT, 64).
32+
%% coordination system Ra parameters
33+
-define(COORD_DEFAULT_WAL_COMPUTE_CHECKSUMS, true).
34+
-define(COORD_DEFAULT_SEGMENT_COMPUTE_CHECKSUMS, true).
35+
-define(COORD_DEFAULT_COMPRESS_MEM_TABLES, true).
3236

3337
-spec setup() -> ok | no_return().
3438

@@ -159,10 +163,40 @@ get_config(coordination = RaSystem) ->
159163
DefaultConfig = ra_system:default_config(),
160164
CoordDataDir = filename:join(
161165
[rabbit:data_dir(), "coordination", node()]),
166+
WalComputeChecksums = application:get_env(rabbit, coordination_wal_compute_checksums,
167+
?COORD_DEFAULT_WAL_COMPUTE_CHECKSUMS),
168+
SegmentComputeChecksums = application:get_env(rabbit, coordination_segment_compute_checksums,
169+
?COORD_DEFAULT_SEGMENT_COMPUTE_CHECKSUMS),
170+
WalMaxSizeBytes = application:get_env(rabbit, coordination_wal_max_size_bytes,
171+
?COORD_WAL_MAX_SIZE_B),
172+
WalMaxEntries = application:get_env(rabbit, coordination_wal_max_entries,
173+
maps:get(wal_max_entries, DefaultConfig)),
174+
WalMaxBatchSize = application:get_env(rabbit, coordination_wal_max_batch_size,
175+
maps:get(wal_max_batch_size, DefaultConfig)),
176+
SegmentMaxSizeBytes = application:get_env(rabbit, coordination_segment_max_size_bytes,
177+
maps:get(segment_max_size_bytes, DefaultConfig)),
178+
SegmentMaxEntries = application:get_env(rabbit, coordination_segment_max_entries,
179+
maps:get(segment_max_entries, DefaultConfig)),
180+
AERBatchSize = application:get_env(rabbit, coordination_max_append_entries_rpc_batch_size,
181+
maps:get(default_max_append_entries_rpc_batch_size, DefaultConfig)),
182+
CompressMemTables = application:get_env(rabbit, coordination_compress_mem_tables,
183+
?COORD_DEFAULT_COMPRESS_MEM_TABLES),
184+
SnapshotChunkSize = application:get_env(rabbit, coordination_snapshot_chunk_size,
185+
maps:get(snapshot_chunk_size, DefaultConfig)),
186+
162187
DefaultConfig#{name => RaSystem,
163188
data_dir => CoordDataDir,
164189
wal_data_dir => CoordDataDir,
165-
wal_max_size_bytes => ?COORD_WAL_MAX_SIZE_B,
190+
wal_compute_checksums => WalComputeChecksums,
191+
segment_compute_checksums => SegmentComputeChecksums,
192+
wal_max_size_bytes => WalMaxSizeBytes,
193+
wal_max_entries => WalMaxEntries,
194+
wal_max_batch_size => WalMaxBatchSize,
195+
segment_max_size_bytes => SegmentMaxSizeBytes,
196+
segment_max_entries => SegmentMaxEntries,
197+
default_max_append_entries_rpc_batch_size => AERBatchSize,
198+
compress_mem_tables => CompressMemTables,
199+
snapshot_chunk_size => SnapshotChunkSize,
166200
names => ra_system:derive_names(RaSystem)}.
167201

168202
-spec ensure_stopped() -> ok | no_return().

‎deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1230,6 +1230,80 @@ credential_validator.regexp = ^abc\\d+",
12301230
]}],
12311231
[]},
12321232

1233+
%%
1234+
%% Coordination System Ra Parameters
1235+
%%
1236+
1237+
{coordination_wal_compute_checksums,
1238+
"coordination.wal_compute_checksums = true",
1239+
[{rabbit, [
1240+
{coordination_wal_compute_checksums, true}
1241+
]}],
1242+
[]},
1243+
1244+
{coordination_segment_compute_checksums,
1245+
"coordination.segment_compute_checksums = true",
1246+
[{rabbit, [
1247+
{coordination_segment_compute_checksums, true}
1248+
]}],
1249+
[]},
1250+
1251+
{coordination_wal_max_size_bytes,
1252+
"coordination.wal_max_size_bytes = 64000000",
1253+
[{rabbit, [
1254+
{coordination_wal_max_size_bytes, 64000000}
1255+
]}],
1256+
[]},
1257+
1258+
{coordination_wal_max_entries,
1259+
"coordination.wal_max_entries = 500000",
1260+
[{rabbit, [
1261+
{coordination_wal_max_entries, 500000}
1262+
]}],
1263+
[]},
1264+
1265+
{coordination_wal_max_batch_size,
1266+
"coordination.wal_max_batch_size = 4096",
1267+
[{rabbit, [
1268+
{coordination_wal_max_batch_size, 4096}
1269+
]}],
1270+
[]},
1271+
1272+
{coordination_segment_max_size_bytes,
1273+
"coordination.segment_max_size_bytes = 64000000",
1274+
[{rabbit, [
1275+
{coordination_segment_max_size_bytes, 64000000}
1276+
]}],
1277+
[]},
1278+
1279+
{coordination_segment_max_entries,
1280+
"coordination.segment_max_entries = 4096",
1281+
[{rabbit, [
1282+
{coordination_segment_max_entries, 4096}
1283+
]}],
1284+
[]},
1285+
1286+
{coordination_max_append_entries_rpc_batch_size,
1287+
"coordination.max_append_entries_rpc_batch_size = 16",
1288+
[{rabbit, [
1289+
{coordination_max_append_entries_rpc_batch_size, 16}
1290+
]}],
1291+
[]},
1292+
1293+
{coordination_compress_mem_tables,
1294+
"coordination.compress_mem_tables = true",
1295+
[{rabbit, [
1296+
{coordination_compress_mem_tables, true}
1297+
]}],
1298+
[]},
1299+
1300+
{coordination_snapshot_chunk_size,
1301+
"coordination.snapshot_chunk_size = 1000000",
1302+
[{rabbit, [
1303+
{coordination_snapshot_chunk_size, 1000000}
1304+
]}],
1305+
[]},
1306+
12331307
%%
12341308
%% Runtime parameters
12351309
%%

‎deps/rabbit/test/rabbit_ra_systems_SUITE.erl‎

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
%% This Source Code Form is subject to the terms of the Mozilla Public
2+
%% License, v. 2.0. If a copy of the MPL was not distributed with this
3+
%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
4+
%%
5+
%% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
6+
%%
7+
8+
-module(rabbit_ra_systems_SUITE).
9+
10+
-compile(export_all).
11+
12+
-include_lib("common_test/include/ct.hrl").
13+
-include_lib("eunit/include/eunit.hrl").
14+
15+
all() ->
16+
[
17+
coordination_defaults,
18+
coordination_wal_max_size_bytes_default,
19+
quorum_queue_defaults
20+
].
21+
22+
init_per_suite(Config) ->
23+
rabbit_ct_helpers:log_environment(),
24+
Config1 = rabbit_ct_helpers:run_setup_steps(Config),
25+
rabbit_ct_config_schema:init_schemas(rabbit, Config1).
26+
27+
end_per_suite(Config) ->
28+
rabbit_ct_helpers:run_teardown_steps(Config).
29+
30+
init_per_testcase(Testcase, Config) ->
31+
rabbit_ct_helpers:testcase_started(Config, Testcase),
32+
Config1 = rabbit_ct_helpers:set_config(Config, [
33+
{rmq_nodename_suffix, Testcase}
34+
]),
35+
rabbit_ct_helpers:run_steps(Config1,
36+
rabbit_ct_broker_helpers:setup_steps() ++
37+
rabbit_ct_client_helpers:setup_steps()).
38+
39+
end_per_testcase(Testcase, Config) ->
40+
Config1 = rabbit_ct_helpers:run_steps(Config,
41+
rabbit_ct_client_helpers:teardown_steps() ++
42+
rabbit_ct_broker_helpers:teardown_steps()),
43+
rabbit_ct_helpers:testcase_finished(Config1, Testcase).
44+
45+
coordination_defaults(Config) ->
46+
ok = rabbit_ct_broker_helpers:rpc(Config, 0,
47+
?MODULE, check_coordination_defaults, []).
48+
49+
coordination_wal_max_size_bytes_default(Config) ->
50+
ok = rabbit_ct_broker_helpers:rpc(Config, 0,
51+
?MODULE, check_coordination_wal_max_size_bytes_default, []).
52+
53+
quorum_queue_defaults(Config) ->
54+
ok = rabbit_ct_broker_helpers:rpc(Config, 0,
55+
?MODULE, check_quorum_queue_defaults, []).
56+
57+
check_coordination_defaults() ->
58+
%% Get the coordination Ra system config
59+
Config = rabbit_ra_systems:get_config(coordination),
60+
61+
%% Verify that all expected keys are present
62+
?assert(maps:is_key(wal_compute_checksums, Config)),
63+
?assert(maps:is_key(segment_compute_checksums, Config)),
64+
?assert(maps:is_key(wal_max_size_bytes, Config)),
65+
?assert(maps:is_key(wal_max_entries, Config)),
66+
?assert(maps:is_key(wal_max_batch_size, Config)),
67+
?assert(maps:is_key(segment_max_size_bytes, Config)),
68+
?assert(maps:is_key(segment_max_entries, Config)),
69+
?assert(maps:is_key(default_max_append_entries_rpc_batch_size, Config)),
70+
?assert(maps:is_key(compress_mem_tables, Config)),
71+
?assert(maps:is_key(snapshot_chunk_size, Config)),
72+
73+
%% Verify default boolean values
74+
?assertEqual(true, maps:get(wal_compute_checksums, Config)),
75+
?assertEqual(true, maps:get(segment_compute_checksums, Config)),
76+
?assertEqual(true, maps:get(compress_mem_tables, Config)),
77+
78+
ok.
79+
80+
check_coordination_wal_max_size_bytes_default() ->
81+
%% Get the coordination Ra system config
82+
Config = rabbit_ra_systems:get_config(coordination),
83+
84+
%% Verify that wal_max_size_bytes defaults to 64 MB (64_000_000 bytes)
85+
WalMaxSizeBytes = maps:get(wal_max_size_bytes, Config),
86+
ExpectedWalMaxSizeBytes = 64_000_000,
87+
?assertEqual(ExpectedWalMaxSizeBytes, WalMaxSizeBytes),
88+
89+
ok.
90+
91+
check_quorum_queue_defaults() ->
92+
%% Get the quorum_queues Ra system config
93+
Config = rabbit_ra_systems:get_config(quorum_queues),
94+
95+
%% Verify that all expected keys are present
96+
?assert(maps:is_key(wal_compute_checksums, Config)),
97+
?assert(maps:is_key(segment_compute_checksums, Config)),
98+
?assert(maps:is_key(wal_max_size_bytes, Config)),
99+
?assert(maps:is_key(wal_max_entries, Config)),
100+
?assert(maps:is_key(wal_max_batch_size, Config)),
101+
?assert(maps:is_key(segment_max_size_bytes, Config)),
102+
?assert(maps:is_key(segment_max_entries, Config)),
103+
?assert(maps:is_key(default_max_append_entries_rpc_batch_size, Config)),
104+
?assert(maps:is_key(compress_mem_tables, Config)),
105+
?assert(maps:is_key(snapshot_chunk_size, Config)),
106+
?assert(maps:is_key(server_recovery_strategy, Config)),
107+
108+
ok.

0 commit comments

Comments
 (0)
⚡