Copilot commented on code in PR #1032: URL: https://github.com/apache/ranger/pull/1032#discussion_r3485965185
########## audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanHolder.java: ########## @@ -0,0 +1,94 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.ranger.audit.producer.kafka.partition; + +import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlan; +import org.apache.ranger.audit.producer.kafka.partition.model.ServiceAllowlistEntry; + +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.Set; +import java.util.concurrent.atomic.AtomicReference; + +/** Hot-path in-memory plan for {@code AuditPartitioner} and the background watcher. */ +public class PartitionPlanHolder { + private static final PartitionPlanHolder INSTANCE = new PartitionPlanHolder(); + + private final AtomicReference<PartitionPlan> planRef = new AtomicReference<>(); + private volatile int lastInstalledVersion; + + private PartitionPlanHolder() { + } + + public static PartitionPlanHolder getInstance() { + return INSTANCE; + } + + public PartitionPlan getPlan() { + return planRef.get(); + } + + public int getLastInstalledVersion() { + return lastInstalledVersion; + } + + /** Validates and atomically installs the plan used by the Kafka partitioner. */ + public void install(PartitionPlan plan, Integer kafkaPartitionCount) { + PartitionPlanValidator.validate(plan, kafkaPartitionCount); + planRef.set(plan); + lastInstalledVersion = plan.getVersion(); + AuthToLocalRuleComposer.getInstance().applyForPlan(plan); + } + + /** + * Returns allowed short usernames for a service repo from the in-memory registry document. + * {@code null} when the plan has no {@code services} block, or when the repo is not present + * in the plan (caller should fall back to static XML). + * Returns an empty set when the repo is present with an explicit empty allowlist (deny all). Review Comment: This Javadoc line claims an explicit empty allowlist is a supported state, but PartitionPlanValidator.validateServices rejects empty allowedUsers. Please align the comment with the actual invariant to avoid confusion. ########## audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/AuditPartitioner.java: ########## @@ -156,6 +174,164 @@ public void close() { appIdCounters.clear(); } + /** + * Routes one audit event to a Kafka partition using the in-memory dynamic partition plan. + * + * <p>The plan splits the audit topic into <em>dedicated plugin lanes</em> (configured plugins + * such as {@code hdfs}, {@code hiveServer}) and a <em>shared buffer pool</em> (everything else). + * Routing for a given {@code appId} follows this order: + * <ol> + * <li><b>Known plugin with dedicated partitions</b> — round-robin across that plugin's + * assignment list so load is spread evenly while preserving per-plugin ordering lanes. + * Example: {@code hdfs} → [0, 1, 2] sends three successive events to 0, then 1, + * then 2, then wraps.</li> + * <li><b>Unknown or unconfigured plugin</b> — sticky hash into the shared buffer pool so + * the same {@code appId} always lands on the same buffer partition. + * Example: buffer → [10, 11]; {@code myCustomApp} consistently maps to 10 or 11.</li> + * <li><b>No buffer partitions in the plan</b> — sticky hash across the full topic when every + * partition is assigned to configured plugins.</li> + * </ol> + * + * <p>After topic scale-up, the Kafka producer's cluster metadata can lag behind + * {@link PartitionPlan#getTopicPartitionCount()}. We use the larger of the two counts as + * {@code effectiveTopicPartitionCount} so a newly planned tail id (e.g. 12) is not folded + * into the stale metadata ceiling (e.g. 11). + * + * @param appId plugin key from the audit event (Kafka record key) + * @param kafkaClusterPartitionCount partition count reported by live Kafka cluster metadata + * @return Kafka partition id to produce to + */ + private int partitionFromPlan(String appId, int kafkaClusterPartitionCount) { + PartitionPlan plan = PartitionPlanHolder.getInstance().getPlan(); + if (plan == null) { + LOG.error("Dynamic partition plan is not loaded; falling back to hash routing for appId '{}'", appId); + return hashAppIdToPartitionIndex(appId, kafkaClusterPartitionCount); + } + + int effectiveTopicPartitionCount = Math.max(kafkaClusterPartitionCount, plan.getTopicPartitionCount()); + + PluginPartitionAssignment pluginAssignment = findPluginAssignment(plan, appId); + if (pluginAssignment != null && !pluginAssignment.getPartitions().isEmpty()) { + List<Integer> dedicatedPluginPartitions = pluginAssignment.getPartitions(); + int dedicatedLaneIndex = nextRoundRobinIndex(appId, dedicatedPluginPartitions.size()); + int plannedPartitionId = dedicatedPluginPartitions.get(dedicatedLaneIndex); + return resolvePlannedPartitionId(plannedPartitionId, effectiveTopicPartitionCount); + } + + List<Integer> sharedBufferPartitions = plan.getBuffer().getPartitions(); + if (sharedBufferPartitions.isEmpty()) { + return hashAppIdToPartitionIndex(appId, effectiveTopicPartitionCount); + } + int bufferPoolIndex = hashAppIdToPartitionIndex(appId, sharedBufferPartitions.size()); + int plannedPartitionId = sharedBufferPartitions.get(bufferPoolIndex); + return resolvePlannedPartitionId(plannedPartitionId, effectiveTopicPartitionCount); + } + + /** + * Sticky hash: same {@code appId} always picks the same slot in {@code [0, slotCount)}. + * Used for buffer-pool routing and for plan-not-loaded fallback. + */ + private static int hashAppIdToPartitionIndex(String appId, int slotCount) { + return Math.abs(appId.hashCode() % slotCount); + } Review Comment: hashAppIdToPartitionIndex() uses Math.abs(appId.hashCode() % slotCount). If appId.hashCode() is Integer.MIN_VALUE, Math.abs returns a negative value, which can produce a negative index (and IndexOutOfBoundsException when indexing the buffer partitions list). Use Math.floorMod to guarantee a non-negative index. ########## audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml: ########## @@ -322,12 +352,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>hdfs,yarn,knox,hive,hiveServer2,hiveMetastore,kafka,hbaseRegional,hbaseMaster,solr,trino,ozone,kudu,nifi</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/services + or promote without editing this list. Review Comment: The configured.plugins description still references the removed onboarding endpoint (POST /api/audit/partition-plan/services). The simplified API in this PR onboards plugins via POST /api/audit/partition-plan/plugins, so this doc example should be updated to match. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
