This is an automated email from the ASF dual-hosted git repository. ramackri pushed a commit to branch ranger-kafka-dynamic-partition-plan in repository https://gitbox.apache.org/repos/asf/ranger.git
commit cfea9ab96ecb38a49447ff49c416b9301a448965 Author: ramk <[email protected]> AuthorDate: Sun Jun 21 14:51:16 2026 +0530 Default configured.plugins to empty and align dynamic bootstrap. Leave kafka.configured.plugins unset in sample XML so greenfield deployments use hash partitioning in static mode and buffer-only plan bootstrap in dynamic mode; operators add plugins via XML or partition-plan REST. Fix empty-registry bootstrap to size the plan from kafka.topic.partitions and update E2E/docs accordingly. Co-authored-by: Cursor <[email protected]> --- .../README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md | 4 +-- .../README-KAFKA-PARTITION-PLAN-E2E-TEST-PLAN.md | 5 ++-- audit-server/README-KAFKA-PLUGIN-PARTITIONING.md | 2 +- .../ranger/audit/server/AuditServerConstants.java | 4 +-- .../kafka/partition/PartitionPlanBootstrap.java | 7 ++++- .../partition/PartitionPlanBootstrapConfig.java | 35 +++++++++++++++------- .../resources/conf/ranger-audit-ingestor-site.xml | 29 +++++++++++++----- .../partition/PartitionPlanBootstrapTest.java | 11 +++++++ .../scripts/audit/partition-plan-e2e-lib.sh | 2 +- .../scripts/audit/verify-partition-plan-e2e.sh | 14 +++++++-- 10 files changed, 85 insertions(+), 28 deletions(-) diff --git a/audit-server/README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md b/audit-server/README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md index 197f7b336..0b397b4bf 100644 --- a/audit-server/README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md +++ b/audit-server/README-KAFKA-DYNAMIC-PARTITIONING-GUIDE.md @@ -298,7 +298,7 @@ When dynamic mode starts and the plan registry is **empty**: 1. Ingestor enables dynamic mode. 2. Watcher finds no document in `ranger_audit_partition_plan`. -3. Ingestor reads XML (`configured.plugins`, overrides, buffer, and `service.*.allowed.users`). +3. Ingestor reads XML (`configured.plugins` if set, else buffer-only bootstrap from `kafka.topic.partitions`; overrides, buffer, and `service.*.allowed.users`). 4. Ingestor builds and publishes the **first document** (`plugins` + `services`) to the compacted topic. 5. Ingestor loads that document into memory and begins enforcing allowlist + routing. @@ -634,7 +634,7 @@ No. The first plan is built from XML (same layout as static mode). Kafka is used Yes — recommended for brownfield clusters. Ingestor will read your pre-loaded plan and will not replace it with a fresh XML bootstrap. **Greenfield vs brownfield cutover — what is different?** -Greenfield: enable dynamic on an empty registry; first pod seeds from XML. Brownfield: export static layout, pre-seed the plan topic (or rely on bootstrap matching XML), then enable dynamic and verify every pod shows the same plan. +Greenfield: enable dynamic on an empty registry; first pod seeds from XML (empty `configured.plugins` → buffer-only plan sized by `kafka.topic.partitions`). Brownfield: export static layout, pre-seed the plan topic (or set `configured.plugins` in XML to match production), then enable dynamic and verify every pod shows the same plan. **Does the `configured.plugins` XML list still matter in dynamic mode?** Yes for **bootstrap only** when the registry is empty. After the first plan exists, runtime routing changes go through REST, not by editing that list. diff --git a/audit-server/README-KAFKA-PARTITION-PLAN-E2E-TEST-PLAN.md b/audit-server/README-KAFKA-PARTITION-PLAN-E2E-TEST-PLAN.md index 4016fa745..1c70a8d2a 100644 --- a/audit-server/README-KAFKA-PARTITION-PLAN-E2E-TEST-PLAN.md +++ b/audit-server/README-KAFKA-PARTITION-PLAN-E2E-TEST-PLAN.md @@ -209,8 +209,9 @@ Unit tests → Tier 3 up → static-only script → enable dynamic → dynamic s | Layout | `configured.plugins` | Buffer | Total partitions | |--------|---------------------|--------|------------------| +| **L0 empty (site XML default)** | *(empty)* | N/A — dynamic bootstrap uses all of `kafka.topic.partitions` as buffer | **`kafka.topic.partitions`** (default **10**) | | L1 minimal | `hdfs` | 3 | 6 | -| L2 standard (site XML default) | `hdfs,yarn,knox,hiveServer2,hiveMetastore,kafka,hbaseRegional,hbaseMaster,solr,trino,ozone,kudu,nifi` (13 plugins) | 9 | **48** (13×3+9) | +| L2 full lab / brownfield | `hdfs,yarn,knox,hiveServer2,hiveMetastore,kafka,hbaseRegional,hbaseMaster,solr,trino,ozone,kudu,nifi` (13 plugins) | 9 | **48** (13×3+9) | | L3 hot | + per-plugin overrides | 9 | varies | -Shipped `ranger-audit-ingestor-site.xml` uses **L2** — do not shorten the plugin list for E2E. Automated promote uses a **buffer** plugin such as **`storm`** (not in `configured.plugins`). +Greenfield E2E uses **L0** unless site XML is patched. Promote tests pick a buffer plugin such as **`storm`** (not in `configured.plugins`). For **L2** layout validation, set the 13-plugin list in site XML before enabling dynamic mode. diff --git a/audit-server/README-KAFKA-PLUGIN-PARTITIONING.md b/audit-server/README-KAFKA-PLUGIN-PARTITIONING.md index 380980493..964a9ce66 100644 --- a/audit-server/README-KAFKA-PLUGIN-PARTITIONING.md +++ b/audit-server/README-KAFKA-PLUGIN-PARTITIONING.md @@ -96,7 +96,7 @@ topic partitions = Σ (partitions per plugin in configured.plugins) + buffer par `ranger.audit.ingestor.kafka.topic.partitions` (default **10** in XML) is used for **hash-based mode only**. In plugin-based mode, topic create/update uses the sum formula above, not `10`. -**Shipped sample XML** lists many plugins in `configured.plugins` (hdfs, yarn, knox, hiveServer2, …). With defaults that yields a **large** topic (e.g. 13 plugins × 3 + 9 buffer = **48**). For a minimal deployment, shorten `configured.plugins` to only plugins you actually use. +**Default site XML** leaves `configured.plugins` **empty** (hash-based static mode; dynamic greenfield bootstraps a **buffer-only** plan using `kafka.topic.partitions`). To pre-assign dedicated ranges at bootstrap, set an explicit comma-separated list (example plugin IDs: hdfs, yarn, knox, hiveServer2, hiveMetastore, kafka, hbaseRegional, hbaseMaster, solr, trino, ozone, kudu, nifi). A full 13-plugin layout yields a large topic (e.g. 13 × 3 + 9 buffer = **48** partitions). ### Hash-based mode (`configured.plugins` empty) diff --git a/audit-server/audit-common/src/main/java/org/apache/ranger/audit/server/AuditServerConstants.java b/audit-server/audit-common/src/main/java/org/apache/ranger/audit/server/AuditServerConstants.java index e3dc9a2ef..cbbdf591b 100644 --- a/audit-server/audit-common/src/main/java/org/apache/ranger/audit/server/AuditServerConstants.java +++ b/audit-server/audit-common/src/main/java/org/apache/ranger/audit/server/AuditServerConstants.java @@ -120,9 +120,9 @@ private AuditServerConstants() {} public static final long DEFAULT_OFFSET_COMMIT_INTERVAL_MS = 30000; // 30 seconds public static final int DEFAULT_MAX_POLL_RECORDS = 500; // Kafka default batch size - // Default configured plugins: each gets allocated partitions from the topic + // Empty by default: operators opt in via XML (static) or REST (dynamic). See ranger-audit-ingestor-site.xml. public static final String DEFAULT_PARTITIONER_CLASS = "org.apache.ranger.audit.producer.kafka.AuditPartitioner"; - public static final String DEFAULT_CONFIGURED_PLUGINS = "hdfs,yarn,knox,hiveServer2,hiveMetastore,kafka,hbaseRegional,hbaseMaster,solr,trino,ozone,kudu,nifi"; + public static final String DEFAULT_CONFIGURED_PLUGINS = ""; public static final short DEFAULT_REPLICATION_FACTOR = 3; public static final int DEFAULT_TOPIC_PARTITIONS = 10; public static final int DEFAULT_PARTITIONS_PER_CONFIGURED_PLUGIN = 3; diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrap.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrap.java index b3aee9777..293d9d1a8 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrap.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrap.java @@ -55,7 +55,12 @@ public static PartitionPlan createInitialPlan(PartitionPlanBootstrapConfig confi nextPartition += count; } - int topicPartitionCount = nextPartition + Math.max(1, config.getBufferPartitionCount()); + int topicPartitionCount; + if (nextPartition == 0) { + topicPartitionCount = config.getHashBasedTopicPartitionCount(); + } else { + topicPartitionCount = nextPartition + Math.max(1, config.getBufferPartitionCount()); + } PartitionPlan plan = PartitionPlan.builder() .topic(config.getAuditTopic()) .version(PartitionPlanConstants.INITIAL_PLAN_VERSION) diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrapConfig.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrapConfig.java index 5ff97b7ff..0e4903ad7 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrapConfig.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrapConfig.java @@ -21,8 +21,10 @@ import org.apache.ranger.audit.server.AuditServerConstants; +import java.util.ArrayList; import java.util.Collections; import java.util.LinkedHashMap; +import java.util.List; import java.util.Map; /** Inputs for building the first partition plan from legacy ingestor XML / producer config. */ @@ -31,24 +33,26 @@ public final class PartitionPlanBootstrapConfig { private final String[] configuredPlugins; private final int defaultPartitionsPerPlugin; private final int bufferPartitionCount; + private final int hashBasedTopicPartitionCount; private final Map<String, Integer> pluginPartitionOverrides; - public PartitionPlanBootstrapConfig(String auditTopic, String[] configuredPlugins, int defaultPartitionsPerPlugin, int bufferPartitionCount, Map<String, Integer> pluginPartitionOverrides) { + public PartitionPlanBootstrapConfig(String auditTopic, String[] configuredPlugins, int defaultPartitionsPerPlugin, int bufferPartitionCount, int hashBasedTopicPartitionCount, Map<String, Integer> pluginPartitionOverrides) { this.auditTopic = auditTopic; this.configuredPlugins = configuredPlugins != null ? configuredPlugins : new String[0]; this.defaultPartitionsPerPlugin = defaultPartitionsPerPlugin; this.bufferPartitionCount = bufferPartitionCount; + this.hashBasedTopicPartitionCount = Math.max(1, hashBasedTopicPartitionCount); this.pluginPartitionOverrides = pluginPartitionOverrides != null ? new LinkedHashMap<>(pluginPartitionOverrides) : Collections.emptyMap(); } public static PartitionPlanBootstrapConfig create(String auditTopic, String[] configuredPlugins, int defaultPartitionsPerPlugin, int bufferPartitionCount) { - return new PartitionPlanBootstrapConfig(auditTopic, configuredPlugins, defaultPartitionsPerPlugin, bufferPartitionCount, Collections.emptyMap()); + return new PartitionPlanBootstrapConfig(auditTopic, configuredPlugins, defaultPartitionsPerPlugin, bufferPartitionCount, AuditServerConstants.DEFAULT_TOPIC_PARTITIONS, Collections.emptyMap()); } public PartitionPlanBootstrapConfig withPluginOverride(String pluginId, int partitionCount) { Map<String, Integer> overrides = new LinkedHashMap<>(pluginPartitionOverrides); overrides.put(pluginId, partitionCount); - return new PartitionPlanBootstrapConfig(auditTopic, configuredPlugins, defaultPartitionsPerPlugin, bufferPartitionCount, overrides); + return new PartitionPlanBootstrapConfig(auditTopic, configuredPlugins, defaultPartitionsPerPlugin, bufferPartitionCount, hashBasedTopicPartitionCount, overrides); } public String getAuditTopic() { @@ -67,6 +71,11 @@ public int getBufferPartitionCount() { return bufferPartitionCount; } + /** Used when {@link #getConfiguredPlugins()} is empty: matches {@code kafka.topic.partitions} / hash-based mode. */ + public int getHashBasedTopicPartitionCount() { + return hashBasedTopicPartitionCount; + } + public int getPartitionsForPlugin(String pluginId) { Integer override = pluginPartitionOverrides.get(pluginId); int count = override != null ? override : defaultPartitionsPerPlugin; @@ -76,11 +85,12 @@ public int getPartitionsForPlugin(String pluginId) { public static PartitionPlanBootstrapConfig fromProducerConfigMap(Map<String, ?> configs, String auditTopic) { String propPrefix = AuditServerConstants.PROP_PREFIX_AUDIT_SERVER; String pluginsStr = getString(configs, propPrefix + AuditServerConstants.PROP_CONFIGURED_PLUGINS, AuditServerConstants.DEFAULT_CONFIGURED_PLUGINS); - String[] plugins = trimPlugins(pluginsStr.split(",")); + String[] plugins = parsePluginIds(pluginsStr); int defaultPerPlugin = getInt(configs, propPrefix + AuditServerConstants.PROP_TOPIC_PARTITIONS_PER_CONFIGURED_PLUGIN, AuditServerConstants.DEFAULT_PARTITIONS_PER_CONFIGURED_PLUGIN); int bufferCount = getInt(configs, propPrefix + AuditServerConstants.PROP_BUFFER_PARTITIONS, AuditServerConstants.DEFAULT_BUFFER_PARTITIONS); + int hashBasedTopicPartitions = getInt(configs, propPrefix + AuditServerConstants.PROP_TOPIC_PARTITIONS, AuditServerConstants.DEFAULT_TOPIC_PARTITIONS); - PartitionPlanBootstrapConfig config = create(auditTopic, plugins, defaultPerPlugin, bufferCount); + PartitionPlanBootstrapConfig config = new PartitionPlanBootstrapConfig(auditTopic, plugins, defaultPerPlugin, bufferCount, hashBasedTopicPartitions, Collections.emptyMap()); for (String plugin : plugins) { String overrideKey = propPrefix + AuditServerConstants.PROP_PLUGIN_PARTITION_OVERRIDE_PREFIX + plugin; if (configs.containsKey(overrideKey)) { @@ -90,12 +100,17 @@ public static PartitionPlanBootstrapConfig fromProducerConfigMap(Map<String, ?> return config; } - private static String[] trimPlugins(String[] plugins) { - String[] trimmed = new String[plugins.length]; - for (int i = 0; i < plugins.length; i++) { - trimmed[i] = plugins[i].trim(); + private static String[] parsePluginIds(String pluginsStr) { + if (pluginsStr == null || pluginsStr.isBlank()) { + return new String[0]; + } + List<String> plugins = new ArrayList<>(); + for (String plugin : pluginsStr.split(",")) { + if (plugin != null && !plugin.isBlank()) { + plugins.add(plugin.trim()); + } } - return trimmed; + return plugins.toArray(new String[0]); } private static String getString(Map<String, ?> configs, String key, String defaultValue) { diff --git a/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml b/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml index fbb58022d..def64c3fe 100644 --- a/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml +++ b/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml @@ -353,12 +353,27 @@ <property> <name>ranger.audit.ingestor.kafka.configured.plugins</name> - <value>hdfs,yarn,knox,hiveServer2,hiveMetastore,kafka,hbaseRegional,hbaseMaster,solr,trino,ozone,kudu,nifi</value> + <value></value> <description> - Comma-separated list of configured plugin IDs. - If set: Uses AuditPartitioner with auto-calculated partitions (sum of plugin allocations + buffer). - If empty/not set: Uses Kafka default hash-based partitioner with topic.partitions value. - Each plugin receives dedicated partitions based on topic.partitions.per.configured.plugin (default: 3). + Comma-separated plugin IDs (agentId / Kafka record key) that receive dedicated partition ranges in static mode. + Leave empty by default; add only the plugins you deploy. + + Empty (recommended greenfield): + - Static mode: Kafka default hash partitioner using kafka.topic.partitions below. + - Dynamic mode (kafka.partition.plan.dynamic.enabled=true): first bootstrap plan uses all + kafka.topic.partitions as buffer; onboard plugins via POST /api/audit/partition-plan/onboard-repo + or promote without editing this list. + + Non-empty (static / bootstrap seed): + - Uses AuditPartitioner: total partitions = sum(per-plugin counts) + kafka.topic.partitions.buffer. + - Append new plugin IDs only; never reorder existing entries (partition ranges are contiguous). + - Dynamic mode: seeds the initial plan from this list when ranger_audit_partition_plan is empty; + runtime changes use partition-plan REST, not XML edits. + + Example plugin IDs (pick what you run): hdfs, yarn, knox, hiveServer2, hiveMetastore, kafka, + hbaseRegional, hbaseMaster, solr, trino, ozone, kudu, nifi. + Per-plugin partition count: kafka.topic.partitions.per.configured.plugin (default 3), overridable via + kafka.plugin.partition.overrides.{pluginId}. </description> </property> @@ -380,8 +395,8 @@ Number of buffer partitions reserved for unconfigured plugins. Used ONLY when configured.plugins is set (for plugin-based partitioning). Total = (sum of plugin partitions) + (buffer partitions). - Example: 7 plugins × 3 + 9 buffer = 30 total. - Ignored when configured.plugins is empty. + Example: 2 plugins × 3 + 9 buffer = 15 total. + Ignored when configured.plugins is empty (hash-based or dynamic buffer-only bootstrap). </description> </property> diff --git a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrapTest.java b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrapTest.java index 4ecd6adc5..84c1a3cee 100644 --- a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrapTest.java +++ b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanBootstrapTest.java @@ -25,6 +25,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Collections; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertIterableEquals; @@ -56,6 +57,16 @@ public void testCreateInitialPlanHonorsPluginOverrides() { assertIterableEquals(List.of(8, 9, 10, 11), plan.getBuffer().getPartitions()); } + @Test + public void testCreateInitialPlanEmptyPluginsUsesHashBasedTopicPartitions() { + PartitionPlan plan = PartitionPlanBootstrap.createInitialPlan( + new PartitionPlanBootstrapConfig(TOPIC, new String[0], 3, 9, 10, Collections.emptyMap())); + + assertEquals(10, plan.getTopicPartitionCount()); + assertEquals(0, plan.getPlugins().size()); + assertIterableEquals(List.of(0, 1, 2, 3, 4, 5, 6, 7, 8, 9), plan.getBuffer().getPartitions()); + } + @Test public void testCreateInitialPlanFromProducerConfig() { String propPrefix = AuditServerConstants.PROP_PREFIX_AUDIT_SERVER; diff --git a/dev-support/ranger-docker/scripts/audit/partition-plan-e2e-lib.sh b/dev-support/ranger-docker/scripts/audit/partition-plan-e2e-lib.sh index 45927e089..ab3035a17 100755 --- a/dev-support/ranger-docker/scripts/audit/partition-plan-e2e-lib.sh +++ b/dev-support/ranger-docker/scripts/audit/partition-plan-e2e-lib.sh @@ -38,7 +38,7 @@ SITE_XMLS=( "/opt/ranger/audit-ingestor/webapp/audit-ingestor/WEB-INF/classes/conf/ranger-audit-ingestor-site.xml" ) -# Must match ranger-audit-ingestor-site.xml (13 plugins × 3 + 9 buffer = 48 partitions). +# Example plugin list for E2E promote tests when site XML leaves configured.plugins empty. PP_CONFIGURED_PLUGINS="hdfs,yarn,knox,hiveServer2,hiveMetastore,kafka,hbaseRegional,hbaseMaster,solr,trino,ozone,kudu,nifi" PP_BUFFER_PROMOTE_CANDIDATES=(storm ambari atlas impala sqoop kylin presto elasticsearch) PP_DEFAULT_PROMOTE_PLUGIN="storm" diff --git a/dev-support/ranger-docker/scripts/audit/verify-partition-plan-e2e.sh b/dev-support/ranger-docker/scripts/audit/verify-partition-plan-e2e.sh index 36f53ccd3..0d91ca012 100755 --- a/dev-support/ranger-docker/scripts/audit/verify-partition-plan-e2e.sh +++ b/dev-support/ranger-docker/scripts/audit/verify-partition-plan-e2e.sh @@ -120,11 +120,21 @@ run_dynamic_tests() { fi plan_plugins="$(printf '%s' "${HTTP_BODY}" | python3 -c "import sys,json; print(len(json.load(sys.stdin).get('plugins') or {}))" 2>/dev/null || echo "")" - expected_plugin_count="$(printf '%s' "${PP_CONFIGURED_PLUGINS}" | python3 -c "import sys; print(len([p for p in sys.stdin.read().split(',') if p.strip()]))")" + local configured_plugins + configured_plugins="$(pp_read_site_prop "${CONTAINER}" "ranger.audit.ingestor.kafka.configured.plugins")" + if [[ -n "${configured_plugins}" ]]; then + expected_plugin_count="$(printf '%s' "${configured_plugins}" | python3 -c "import sys; print(len([p for p in sys.stdin.read().split(',') if p.strip()]))")" + else + expected_plugin_count="0" + fi topic_count="$(pp_json_field "${HTTP_BODY}" topicPartitionCount)" kafka_parts="$(pp_kafka_topic_partition_count "${AUDIT_TOPIC}")" if [[ "${version}" == "1" && "${plan_plugins}" == "${expected_plugin_count}" ]]; then - pp_record_pass "greenfield plan lists ${plan_plugins} configured plugins (site XML layout)" + if [[ "${expected_plugin_count}" == "0" ]]; then + pp_record_pass "greenfield plan is buffer-only (configured.plugins empty in site XML)" + else + pp_record_pass "greenfield plan lists ${plan_plugins} configured plugins (site XML layout)" + fi elif [[ "${version}" != "1" ]]; then pp_record_pass "plan has ${plan_plugins} plugins at v${version}" else
