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 6a0971a75a0e2700cf016a535af6811050053c65 Author: ramk <[email protected]> AuthorDate: Tue Jun 23 18:45:27 2026 +0530 Clarify dynamic partition routing in AuditPartitioner for reviewers. Document partitionFromPlan with examples, rename variables to reflect dedicated plugin lanes vs shared buffer pool, improve empty-partition log messages, and rename boundPartitionToTopic to resolvePlannedPartitionId. Co-authored-by: Cursor <[email protected]> --- .../audit/producer/kafka/AuditPartitioner.java | 105 ++++++++++++++------- 1 file changed, 73 insertions(+), 32 deletions(-) diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/AuditPartitioner.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/AuditPartitioner.java index af825e14f..f9a0c8373 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/AuditPartitioner.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/AuditPartitioner.java @@ -174,42 +174,75 @@ public void close() { appIdCounters.clear(); } - /** Picks a partition from the in-memory plan: round-robin for known plugins, hash for buffer. */ - private int partitionFromPlan(String appId, int clusterPartitionCount) { + /** + * 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, clusterPartitionCount); + return hashAppIdToPartitionIndex(appId, kafkaClusterPartitionCount); } - // After scale, producer cluster metadata can lag behind plan.topicPartitionCount; trust the plan - // so planned tail ids (e.g. 12) are not clamped into buffer (e.g. 11). - int routingPartitionCount = Math.max(clusterPartitionCount, plan.getTopicPartitionCount()); - PluginPartitionAssignment assignment = findPluginAssignment(plan, appId); - if (assignment != null && !assignment.getPartitions().isEmpty()) { - List<Integer> partitionIds = assignment.getPartitions(); - int roundRobinIndex = nextRoundRobinIndex(appId, partitionIds.size()); - return boundPartitionToTopic(partitionIds.get(roundRobinIndex), routingPartitionCount); + + 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> bufferIds = plan.getBuffer().getPartitions(); - if (bufferIds.isEmpty()) { - return hashAppIdToPartitionIndex(appId, routingPartitionCount); + + List<Integer> sharedBufferPartitions = plan.getBuffer().getPartitions(); + if (sharedBufferPartitions.isEmpty()) { + return hashAppIdToPartitionIndex(appId, effectiveTopicPartitionCount); } - int bufferIndex = hashAppIdToPartitionIndex(appId, bufferIds.size()); - return boundPartitionToTopic(bufferIds.get(bufferIndex), routingPartitionCount); + int bufferPoolIndex = hashAppIdToPartitionIndex(appId, sharedBufferPartitions.size()); + int plannedPartitionId = sharedBufferPartitions.get(bufferPoolIndex); + return resolvePlannedPartitionId(plannedPartitionId, effectiveTopicPartitionCount); } - /** Sticky hash: maps {@code appId} to an index in {@code [0, partitionCount)}. */ - private static int hashAppIdToPartitionIndex(String appId, int partitionCount) { - return Math.abs(appId.hashCode() % partitionCount); + /** + * 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); } /** - * Picks the next slot for round-robin routing for one plugin. - * Each {@code appId} keeps its own counter (0, 1, 2, …); {@code % laneCount} cycles through lanes. + * Returns the next dedicated-lane index for round-robin routing within one plugin's partition set. + * Each {@code appId} keeps its own counter (0, 1, 2, …); {@code % dedicatedLaneCount} cycles + * through that plugin's lanes only. */ - private int nextRoundRobinIndex(String appId, int laneCount) { + private int nextRoundRobinIndex(String appId, int dedicatedLaneCount) { AtomicInteger messageCounter = appIdCounters.computeIfAbsent(appId, k -> new AtomicInteger(0)); - return messageCounter.getAndIncrement() % laneCount; + return messageCounter.getAndIncrement() % dedicatedLaneCount; } /** Looks up a plugin assignment using case-insensitive plugin id matching. */ @@ -232,17 +265,23 @@ private static int resolveTopicPartitionCount(Cluster cluster, String topic) { } /** - * Returns a valid partition id for routing. Planned ids below {@code routingPartitionCount} are returned - * as-is; never clamp a tail id into the previous last partition when cluster metadata is stale. + * Converts a partition id from the plan into the id passed to the Kafka producer. + * + * <p>When cluster metadata is stale after scale-up, {@code effectiveTopicPartitionCount} may + * exceed what the broker metadata reports. Planned tail ids must still be returned as-is — + * never clamp partition 12 down to 11 just because metadata has not caught up yet. + * + * @param plannedPartitionId partition id from the dynamic plan assignment or buffer pool + * @param effectiveTopicPartitionCount {@code max(kafkaClusterPartitionCount, plan.topicPartitionCount)} */ - private static int boundPartitionToTopic(int plannedPartitionId, int routingPartitionCount) { - if (routingPartitionCount <= 0) { + private static int resolvePlannedPartitionId(int plannedPartitionId, int effectiveTopicPartitionCount) { + if (effectiveTopicPartitionCount <= 0) { return 0; } if (plannedPartitionId < 0) { return 0; } - if (plannedPartitionId < routingPartitionCount) { + if (plannedPartitionId < effectiveTopicPartitionCount) { return plannedPartitionId; } return plannedPartitionId; @@ -279,16 +318,18 @@ private void logDynamicPlanConfiguration(Map<String, ?> configs, String propPref private static String formatPartitionRangeInfo(List<Integer> partitionIds) { if (partitionIds.isEmpty()) { - return "0 partitions (range: n/a): "; + return "no partitions assigned in plan"; } - return String.format("%d partitions (range: %d-%d): ", partitionIds.size(), partitionIds.get(0), partitionIds.get(partitionIds.size() - 1)); + return String.format("%d partitions (range: %d-%d)", + partitionIds.size(), partitionIds.get(0), partitionIds.get(partitionIds.size() - 1)); } private static String formatBufferPartitionInfo(List<Integer> bufferPartitionIds) { if (bufferPartitionIds.isEmpty()) { - return "0 partitions (range: n/a)"; + return "none (all topic partitions assigned to configured plugins)"; } - return String.format("%d partitions (range: %d-%d)", bufferPartitionIds.size(), bufferPartitionIds.get(0), bufferPartitionIds.get(bufferPartitionIds.size() - 1)); + return String.format("%d partitions (range: %d-%d)", + bufferPartitionIds.size(), bufferPartitionIds.get(0), bufferPartitionIds.get(bufferPartitionIds.size() - 1)); } private int indexOfConfiguredPlugin(String appId) {
