From 0ced75348c1bacf69200ea409d2d16704f252abc Mon Sep 17 00:00:00 2001 From: ramk Date: Thu, 6 Aug 2026 08:44:01 +0530 Subject: [PATCH 1/4] RANGER-5719: Shared partition-plan library for admin-managed audit routing Add agents-common partition plan model, allocator, validator, routing helpers, and PolicyDownloadAuthUsersUtil for RANGER-5655. SPIFFE header utilities move to RANGER-5723 (#1139). --- .../AuditPartitionPlanAdminConfig.java | 40 +++ .../AuditPartitionPlanConstants.java | 37 +++ .../partition/PartitionPlanAllocator.java | 297 ++++++++++++++++++ .../partition/PartitionPlanRoutingUtils.java | 70 +++++ .../partition/PartitionPlanValidator.java | 144 +++++++++ .../PolicyDownloadAuthUsersUtil.java | 96 ++++++ .../exception/PartitionPlanException.java | 32 ++ .../audit/partition/model/BufferEntry.java | 73 +++++ .../audit/partition/model/PartitionPlan.java | 260 +++++++++++++++ .../audit/partition/model/PluginEntry.java | 122 +++++++ .../AuditPartitionPlanAdminConfigTest.java | 47 +++ .../partition/PartitionPlanAllocatorTest.java | 164 ++++++++++ .../PartitionPlanRoutingUtilsTest.java | 43 +++ .../partition/PartitionPlanTestSupport.java | 62 ++++ .../partition/PartitionPlanValidatorTest.java | 154 +++++++++ .../PolicyDownloadAuthUsersUtilTest.java | 75 +++++ .../model/PartitionPlanJsonTest.java | 73 +++++ 17 files changed, 1789 insertions(+) create mode 100644 agents-common/src/main/java/org/apache/ranger/audit/partition/AuditPartitionPlanAdminConfig.java create mode 100644 agents-common/src/main/java/org/apache/ranger/audit/partition/AuditPartitionPlanConstants.java create mode 100644 agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanAllocator.java create mode 100644 agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanRoutingUtils.java create mode 100644 agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanValidator.java create mode 100644 agents-common/src/main/java/org/apache/ranger/audit/partition/PolicyDownloadAuthUsersUtil.java create mode 100644 agents-common/src/main/java/org/apache/ranger/audit/partition/exception/PartitionPlanException.java create mode 100644 agents-common/src/main/java/org/apache/ranger/audit/partition/model/BufferEntry.java create mode 100644 agents-common/src/main/java/org/apache/ranger/audit/partition/model/PartitionPlan.java create mode 100644 agents-common/src/main/java/org/apache/ranger/audit/partition/model/PluginEntry.java create mode 100644 agents-common/src/test/java/org/apache/ranger/audit/partition/AuditPartitionPlanAdminConfigTest.java create mode 100644 agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanAllocatorTest.java create mode 100644 agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanRoutingUtilsTest.java create mode 100644 agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanTestSupport.java create mode 100644 agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanValidatorTest.java create mode 100644 agents-common/src/test/java/org/apache/ranger/audit/partition/PolicyDownloadAuthUsersUtilTest.java create mode 100644 agents-common/src/test/java/org/apache/ranger/audit/partition/model/PartitionPlanJsonTest.java diff --git a/agents-common/src/main/java/org/apache/ranger/audit/partition/AuditPartitionPlanAdminConfig.java b/agents-common/src/main/java/org/apache/ranger/audit/partition/AuditPartitionPlanAdminConfig.java new file mode 100644 index 00000000000..67e1c4518c3 --- /dev/null +++ b/agents-common/src/main/java/org/apache/ranger/audit/partition/AuditPartitionPlanAdminConfig.java @@ -0,0 +1,40 @@ +/* + * 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.partition; + +import org.apache.commons.lang3.StringUtils; +import org.apache.hadoop.conf.Configuration; + +/** Reads Admin site configuration for audit partition plan allocation. */ +public final class AuditPartitionPlanAdminConfig { + private AuditPartitionPlanAdminConfig() { + } + + public static int resolvePartitionsPerPlugin(String pluginId, Configuration config) { + if (config == null || StringUtils.isBlank(pluginId)) { + return AuditPartitionPlanConstants.DEFAULT_PARTITIONS_PER_PLUGIN; + } + String overrideKey = AuditPartitionPlanConstants.PROP_ADMIN_PLUGIN_PARTITION_OVERRIDE_PREFIX + pluginId.trim(); + if (StringUtils.isNotBlank(config.get(overrideKey))) { + return config.getInt(overrideKey, AuditPartitionPlanConstants.DEFAULT_PARTITIONS_PER_PLUGIN); + } + return config.getInt(AuditPartitionPlanConstants.PROP_ADMIN_PARTITIONS_PER_PLUGIN, AuditPartitionPlanConstants.DEFAULT_PARTITIONS_PER_PLUGIN); + } +} diff --git a/agents-common/src/main/java/org/apache/ranger/audit/partition/AuditPartitionPlanConstants.java b/agents-common/src/main/java/org/apache/ranger/audit/partition/AuditPartitionPlanConstants.java new file mode 100644 index 00000000000..38c20c6547f --- /dev/null +++ b/agents-common/src/main/java/org/apache/ranger/audit/partition/AuditPartitionPlanConstants.java @@ -0,0 +1,37 @@ +/* + * 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.partition; + +public final class AuditPartitionPlanConstants { + public static final int INITIAL_PLAN_VERSION = 1; + public static final String DEFAULT_AUDIT_TOPIC = "ranger_audits"; + + /** Default partition slots allocated when a plugin is first promoted from buffer. */ + public static final int DEFAULT_PARTITIONS_PER_PLUGIN = 3; + + /** Admin site: {@code ranger-admin-default-site.xml} / {@code ranger-admin-site.xml}. */ + public static final String PROP_ADMIN_PARTITIONS_PER_PLUGIN = "ranger.admin.audit.partition.plan.partitions.per.plugin"; + + /** Per-plugin override prefix, e.g. {@code ...plugin.partition.overrides.hiveServer2}. */ + public static final String PROP_ADMIN_PLUGIN_PARTITION_OVERRIDE_PREFIX = "ranger.admin.audit.partition.plan.plugin.partition.overrides."; + + private AuditPartitionPlanConstants() { + } +} diff --git a/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanAllocator.java b/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanAllocator.java new file mode 100644 index 00000000000..a8507d2e115 --- /dev/null +++ b/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanAllocator.java @@ -0,0 +1,297 @@ +/* + * 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.partition; + +import org.apache.commons.lang3.StringUtils; +import org.apache.ranger.audit.partition.exception.PartitionPlanException; +import org.apache.ranger.audit.partition.model.BufferEntry; +import org.apache.ranger.audit.partition.model.PartitionPlan; +import org.apache.ranger.audit.partition.model.PluginEntry; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; + +import static java.util.Objects.requireNonNull; + +/** Append-only plan updates for Admin-managed audit partition routing. */ +public final class PartitionPlanAllocator { + private PartitionPlanAllocator() { + } + + /** + * Onboard a Ranger service repo under a plugin type. Promotes the plugin from buffer when needed, + * otherwise adds the service to an existing plugin entry. + */ + public static PartitionPlan onboardService(PartitionPlan current, String pluginId, String serviceName, int partitionCount, String updatedBy) { + requireMutationInputs(current, pluginId, partitionCount, updatedBy); + if (StringUtils.isBlank(serviceName)) { + throw new PartitionPlanException("serviceName is required"); + } + String trimmedService = serviceName.trim(); + PluginEntry existing = current.getPlugins().get(pluginId); + if (existing != null) { + return addServiceToPlugin(current, pluginId, trimmedService, updatedBy); + } + return promotePlugin(current, pluginId, partitionCount, updatedBy, trimmedService); + } + + /** Adds a service repo to an already-promoted plugin without changing partition assignment. */ + public static PartitionPlan addServiceToPlugin(PartitionPlan current, String pluginId, String serviceName, String updatedBy) { + if (current == null) { + throw new PartitionPlanException("Current plan is required"); + } + PartitionPlanValidator.validate(current); + if (StringUtils.isBlank(pluginId) || StringUtils.isBlank(serviceName) || StringUtils.isBlank(updatedBy)) { + throw new PartitionPlanException("pluginId, serviceName, and updatedBy are required"); + } + PluginEntry existing = current.getPlugins().get(pluginId); + if (existing == null) { + throw new PartitionPlanException("Plugin '" + pluginId + "' is not configured; promote it first"); + } + String trimmedService = serviceName.trim(); + if (existing.getServices().contains(trimmedService)) { + return current; + } + ensureServiceNotAssignedElsewhere(current.getPlugins(), pluginId, trimmedService); + + Map plugins = new LinkedHashMap<>(current.getPlugins()); + plugins.put(pluginId, existing.addService(trimmedService)); + return commitPlanUpdate(current, updatedBy, current.getTopicPartitionCount(), plugins, current.getBuffer().getPartitions()); + } + + /** Removes a service repo from whichever plugin currently owns it. */ + public static PartitionPlan removeService(PartitionPlan current, String serviceName, String updatedBy) { + if (current == null) { + throw new PartitionPlanException("Current plan is required"); + } + PartitionPlanValidator.validate(current); + if (StringUtils.isBlank(serviceName) || StringUtils.isBlank(updatedBy)) { + throw new PartitionPlanException("serviceName and updatedBy are required"); + } + String trimmedService = serviceName.trim(); + String owningPluginId = findPluginForService(current.getPlugins(), trimmedService); + if (owningPluginId == null) { + return current; + } + + PluginEntry existing = requireNonNull(current.getPlugins().get(owningPluginId)); + List remainingServices = new ArrayList<>(existing.getServices()); + remainingServices.remove(trimmedService); + + Map plugins = new LinkedHashMap<>(current.getPlugins()); + plugins.put(owningPluginId, existing.withServices(remainingServices)); + return commitPlanUpdate(current, updatedBy, current.getTopicPartitionCount(), plugins, current.getBuffer().getPartitions()); + } + + public static PartitionPlan promotePlugin(PartitionPlan current, String pluginId, int partitionCount, String updatedBy) { + return promotePlugin(current, pluginId, partitionCount, updatedBy, null); + } + + /** + * Give a plugin its own partitions. Uses buffer IDs first; adds new tail IDs when buffer is too small. + * Optionally attaches {@code serviceName} to the new plugin entry. + */ + public static PartitionPlan promotePlugin(PartitionPlan current, String pluginId, int partitionCount, String updatedBy, String serviceName) { + requireMutationInputs(current, pluginId, partitionCount, updatedBy); + if (current.getPlugins().containsKey(pluginId)) { + assertPromoteNotConflicting(current, pluginId, partitionCount, serviceName); + throw new PartitionPlanException("Plugin '" + pluginId + "' already has dedicated partitions"); + } + if (StringUtils.isNotBlank(serviceName)) { + ensureServiceNotAssignedElsewhere(current.getPlugins(), pluginId, serviceName.trim()); + } + + List remainingBuffer = new ArrayList<>(current.getBuffer().getPartitions()); + List newPluginIds = takeFromBuffer(remainingBuffer, partitionCount); + int topicPartitionCount = current.getTopicPartitionCount(); + int additionalNeeded = partitionCount - newPluginIds.size(); + if (additionalNeeded > 0) { + topicPartitionCount = appendTailPartitions(newPluginIds, topicPartitionCount, additionalNeeded, collectAssignedPartitionIds(current)); + } + + List services = StringUtils.isNotBlank(serviceName) ? List.of(serviceName.trim()) : List.of(); + Map plugins = addPluginAssignment(current, pluginId, newPluginIds, services); + return commitPlanUpdate(current, updatedBy, topicPartitionCount, plugins, remainingBuffer); + } + + /** Add more partitions to an existing plugin by appending new tail IDs only. */ + public static PartitionPlan scalePlugin(PartitionPlan current, String pluginId, int additionalPartitions, String updatedBy) { + requireMutationInputs(current, pluginId, additionalPartitions, updatedBy); + if (!current.getPlugins().containsKey(pluginId)) { + throw new PartitionPlanException("Plugin '" + pluginId + "' is not configured; promote it first"); + } + + List pluginIds = new ArrayList<>(current.getPlugins().get(pluginId).getPartitions()); + int topicPartitionCount = appendTailPartitions(pluginIds, current.getTopicPartitionCount(), additionalPartitions, collectAssignedPartitionIds(current)); + + Map plugins = addPluginAssignment(current, pluginId, pluginIds, current.getPlugins().get(pluginId).getServices()); + return commitPlanUpdate(current, updatedBy, topicPartitionCount, plugins, current.getBuffer().getPartitions()); + } + + public static boolean isOnboardAlreadyApplied(PartitionPlan current, String pluginId, String serviceName, int partitionCount) { + if (current == null || StringUtils.isBlank(serviceName)) { + return false; + } + PluginEntry existing = current.getPlugins().get(pluginId); + if (existing == null) { + return false; + } + return existing.getPartitions().size() == partitionCount && existing.getServices().contains(serviceName.trim()); + } + + public static boolean isPromoteAlreadyApplied(PartitionPlan current, String pluginId, int partitionCount) { + if (current == null) { + return false; + } + PluginEntry existing = current.getPlugins().get(pluginId); + return existing != null && existing.getPartitions().size() == partitionCount; + } + + /** Applies a merged plan with append-only checks against the current plan. */ + public static PartitionPlan replacePlan(PartitionPlan current, PartitionPlan proposed) { + if (current == null || proposed == null) { + throw new PartitionPlanException("Current and proposed plans are required"); + } + if (!StringUtils.equals(current.getTopic(), proposed.getTopic())) { + throw new PartitionPlanException("Proposed topic must match current topic"); + } + PartitionPlan next = proposed.toBuilder().version(current.getVersion() + 1).build(); + PartitionPlanValidator.validate(next); + PartitionPlanValidator.validateAppendOnly(current, next); + return next; + } + + /** Updates audit POST allow-list metadata; bumps version only when the map changes. */ + public static PartitionPlan updateServiceAllowedUsers(PartitionPlan current, Map> serviceAllowedUsers, String updatedBy) { + if (current == null) { + throw new PartitionPlanException("Current plan is required"); + } + PartitionPlanValidator.validate(current); + if (StringUtils.isBlank(updatedBy)) { + throw new PartitionPlanException("updatedBy is required"); + } + + Map> normalized = PolicyDownloadAuthUsersUtil.normalizeServiceAllowedUsers(serviceAllowedUsers); + if (Objects.equals(current.getServiceAllowedUsers(), normalized)) { + return current; + } + + PartitionPlan next = current.toBuilder() + .version(current.getVersion() + 1) + .serviceAllowedUsers(normalized) + .updatedAt(Instant.now().toString()) + .updatedBy(updatedBy) + .build(); + PartitionPlanValidator.validate(next); + PartitionPlanValidator.validateAppendOnly(current, next); + return next; + } + + private static List takeFromBuffer(List bufferIds, int count) { + List taken = new ArrayList<>(Math.min(count, bufferIds.size())); + while (taken.size() < count && !bufferIds.isEmpty()) { + taken.add(bufferIds.remove(0)); + } + return taken; + } + + private static int appendTailPartitions(List target, int topicPartitionCount, int count, Set assigned) { + int nextId = assigned.isEmpty() ? 1 : assigned.stream().mapToInt(Integer::intValue).max().orElse(0) + 1; + for (int i = 0; i < count; i++) { + target.add(nextId++); + assigned.add(target.get(target.size() - 1)); + } + return topicPartitionCount + count; + } + + private static Set collectAssignedPartitionIds(PartitionPlan plan) { + Set assigned = new HashSet<>(plan.getBuffer().getPartitions()); + for (PluginEntry entry : plan.getPlugins().values()) { + assigned.addAll(entry.getPartitions()); + } + return assigned; + } + + private static Map addPluginAssignment(PartitionPlan current, String pluginId, List partitionIds, List services) { + Map plugins = new LinkedHashMap<>(current.getPlugins()); + plugins.put(pluginId, new PluginEntry(partitionIds, services)); + return plugins; + } + + private static PartitionPlan commitPlanUpdate(PartitionPlan current, String updatedBy, int topicPartitionCount, Map plugins, List bufferIds) { + PartitionPlan next = current.toBuilder() + .version(current.getVersion() + 1) + .topicPartitionCount(topicPartitionCount) + .plugins(plugins) + .buffer(new BufferEntry(bufferIds)) + .updatedAt(Instant.now().toString()) + .updatedBy(updatedBy) + .build(); + PartitionPlanValidator.validate(next); + PartitionPlanValidator.validateAppendOnly(current, next); + return next; + } + + private static String findPluginForService(Map plugins, String serviceName) { + for (Map.Entry entry : plugins.entrySet()) { + if (entry.getValue().getServices().contains(serviceName)) { + return entry.getKey(); + } + } + return null; + } + + private static void ensureServiceNotAssignedElsewhere(Map plugins, String pluginId, String serviceName) { + for (Map.Entry entry : plugins.entrySet()) { + if (!entry.getKey().equals(pluginId) && entry.getValue().getServices().contains(serviceName)) { + throw new PartitionPlanException("Service '" + serviceName + "' is already assigned to plugin '" + entry.getKey() + "'"); + } + } + } + + private static void assertPromoteNotConflicting(PartitionPlan current, String pluginId, int partitionCount, String serviceName) { + PluginEntry existing = requireNonNull(current.getPlugins().get(pluginId)); + if (existing.getPartitions().size() != partitionCount) { + throw new PartitionPlanException("Plugin '" + pluginId + "' already has " + existing.getPartitions().size() + " dedicated partition(s); requested " + partitionCount); + } + if (StringUtils.isNotBlank(serviceName) && existing.getServices().contains(serviceName.trim())) { + return; + } + if (StringUtils.isNotBlank(serviceName)) { + throw new PartitionPlanException("Plugin '" + pluginId + "' already has dedicated partitions"); + } + } + + private static void requireMutationInputs(PartitionPlan current, String pluginId, int partitionCount, String updatedBy) { + if (current == null) { + throw new PartitionPlanException("Current plan is required"); + } + PartitionPlanValidator.validate(current); + if (StringUtils.isBlank(pluginId) || partitionCount < 1 || StringUtils.isBlank(updatedBy)) { + throw new PartitionPlanException("pluginId, partitionCount, and updatedBy are required"); + } + } +} diff --git a/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanRoutingUtils.java b/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanRoutingUtils.java new file mode 100644 index 00000000000..65849f756b5 --- /dev/null +++ b/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanRoutingUtils.java @@ -0,0 +1,70 @@ +/* + * 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.partition; + +import org.apache.commons.lang3.StringUtils; + +/** Converts Admin-managed plan partition ids to Kafka producer partition indices. */ +public final class PartitionPlanRoutingUtils { + private PartitionPlanRoutingUtils() { + } + + /** + * Plan partition ids are 1-based logical ids ({@code 1..topicPartitionCount}). + * Kafka partition indices are 0-based. + */ + public static int toKafkaPartitionIndex(int plannedPartitionId) { + if (plannedPartitionId < 1) { + return 0; + } + return plannedPartitionId - 1; + } + + /** + * Returns a non-negative slot index in {@code [0, slotCount)} for hash-based buffer routing. + * Uses {@link Math#floorMod(int, int)} so {@code Integer.MIN_VALUE} hash codes are safe. + */ + public static int hashToSlotIndex(String key, int slotCount) { + if (slotCount <= 0) { + return 0; + } + if (StringUtils.isBlank(key)) { + return 0; + } + return Math.floorMod(key.hashCode(), slotCount); + } + + /** + * Returns the Kafka partition index for a planned id, clamped to the effective topic size when metadata lags. + */ + public static int resolveKafkaPartitionIndex(int plannedPartitionId, int effectiveTopicPartitionCount) { + if (effectiveTopicPartitionCount <= 0) { + return 0; + } + int kafkaIndex = toKafkaPartitionIndex(plannedPartitionId); + if (kafkaIndex < 0) { + return 0; + } + if (kafkaIndex >= effectiveTopicPartitionCount) { + return effectiveTopicPartitionCount - 1; + } + return kafkaIndex; + } +} diff --git a/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanValidator.java b/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanValidator.java new file mode 100644 index 00000000000..79b4dc95f1b --- /dev/null +++ b/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanValidator.java @@ -0,0 +1,144 @@ +/* + * 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.partition; + +import org.apache.commons.lang3.StringUtils; +import org.apache.ranger.audit.partition.exception.PartitionPlanException; +import org.apache.ranger.audit.partition.model.PartitionPlan; +import org.apache.ranger.audit.partition.model.PluginEntry; + +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Checks partition plan shape and append-only updates. */ +public final class PartitionPlanValidator { + private PartitionPlanValidator() { + } + + public static void validate(PartitionPlan plan) { + validate(plan, null); + } + + /** + * When kafkaPartitionCount is set, it must be at least plan.topicPartitionCount. + * Extra live Kafka partitions (e.g. after static-mode migration) are allowed. + */ + public static void validate(PartitionPlan plan, Integer kafkaPartitionCount) { + if (plan == null || StringUtils.isBlank(plan.getTopic()) || plan.getVersion() < AuditPartitionPlanConstants.INITIAL_PLAN_VERSION || plan.getTopicPartitionCount() < 1) { + throw new PartitionPlanException("Invalid partition plan"); + } + if (kafkaPartitionCount != null && kafkaPartitionCount < plan.getTopicPartitionCount()) { + throw new PartitionPlanException("Kafka topic has fewer partitions than plan requires"); + } + + Set assigned = new HashSet<>(); + registerPartitions(plan.getBuffer().getPartitions(), assigned, true); + for (Map.Entry entry : plan.getPlugins().entrySet()) { + if (StringUtils.isBlank(entry.getKey())) { + throw new PartitionPlanException("Plugin id is required"); + } + registerPartitions(entry.getValue().getPartitions(), assigned, false); + } + if (assigned.size() != plan.getTopicPartitionCount()) { + throw new PartitionPlanException("topicPartitionCount must equal the union of all assigned partition ids"); + } + validateServiceUniqueness(plan.getPlugins()); + validateServiceAllowedUsers(plan.getServiceAllowedUsers()); + } + + /** + * When a service repo is listed in {@code serviceAllowedUsers}, it must have at least one + * allowed short username (from Admin {@code policy.download.auth.users}). + */ + public static void validateServiceAllowedUsers(Map> serviceAllowedUsers) { + if (serviceAllowedUsers == null || serviceAllowedUsers.isEmpty()) { + return; + } + for (Map.Entry> entry : serviceAllowedUsers.entrySet()) { + if (StringUtils.isBlank(entry.getKey())) { + throw new PartitionPlanException("Service repo name is required"); + } + List users = entry.getValue(); + if (users == null || users.isEmpty()) { + throw new PartitionPlanException( + "allowedUsers must not be empty for service '" + entry.getKey().trim() + "'"); + } + } + } + + /** Each Ranger service repo name may appear in at most one plugin entry. */ + public static void validateServiceUniqueness(Map plugins) { + if (plugins == null || plugins.isEmpty()) { + return; + } + Set seenServices = new HashSet<>(); + for (Map.Entry entry : plugins.entrySet()) { + for (String serviceName : entry.getValue().getServices()) { + if (!seenServices.add(serviceName)) { + throw new PartitionPlanException("Service '" + serviceName + "' is assigned to more than one plugin"); + } + } + } + } + + /** New plan must only add tail partitions; existing plugin lists stay unchanged in order. */ + public static void validateAppendOnly(PartitionPlan current, PartitionPlan proposed) { + if (current == null || proposed == null) { + throw new PartitionPlanException("Current and proposed plans are required"); + } + if (proposed.getTopicPartitionCount() < current.getTopicPartitionCount() || proposed.getVersion() != current.getVersion() + 1) { + throw new PartitionPlanException("Plan must grow partition count and increment version by one"); + } + + for (Map.Entry entry : current.getPlugins().entrySet()) { + String pluginId = entry.getKey(); + List before = entry.getValue().getPartitions(); + PluginEntry afterEntry = proposed.getPlugins().get(pluginId); + if (afterEntry == null) { + throw new PartitionPlanException("Append-only violation for plugin '" + pluginId + "'"); + } + List after = afterEntry.getPartitions(); + if (after.size() < before.size()) { + throw new PartitionPlanException("Append-only violation for plugin '" + pluginId + "'"); + } + for (int i = 0; i < before.size(); i++) { + if (!before.get(i).equals(after.get(i))) { + throw new PartitionPlanException("Append-only violation for plugin '" + pluginId + "' at index " + i); + } + } + } + } + + private static void registerPartitions(List partitionIds, Set assigned, boolean allowEmpty) { + if (partitionIds.isEmpty()) { + if (allowEmpty) { + return; + } + throw new PartitionPlanException("Plugin partition list must not be empty"); + } + for (int partitionId : partitionIds) { + if (partitionId < 1 || !assigned.add(partitionId)) { + throw new PartitionPlanException("Invalid or duplicate partition id: " + partitionId); + } + } + } +} diff --git a/agents-common/src/main/java/org/apache/ranger/audit/partition/PolicyDownloadAuthUsersUtil.java b/agents-common/src/main/java/org/apache/ranger/audit/partition/PolicyDownloadAuthUsersUtil.java new file mode 100644 index 00000000000..08d6844f288 --- /dev/null +++ b/agents-common/src/main/java/org/apache/ranger/audit/partition/PolicyDownloadAuthUsersUtil.java @@ -0,0 +1,96 @@ +/* + * 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.partition; + +import org.apache.commons.lang3.StringUtils; +import org.apache.ranger.plugin.model.RangerService; + +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +/** Parses {@code policy.download.auth.users} service config for audit ingestor authorization. */ +public final class PolicyDownloadAuthUsersUtil { + public static final String CONFIG_NAME = "policy.download.auth.users"; + + /** + * Ingestor site key pattern: {@code ranger.audit.ingestor.service..allowed.users}. + * Values originate from Admin {@link #CONFIG_NAME}; partition plan {@code serviceAllowedUsers} + * keys use the same {@code } names (Policy Manager service name). + */ + public static final String INGESTOR_ALLOWED_USERS_SUFFIX = "allowed.users"; + + private PolicyDownloadAuthUsersUtil() { + } + + public static List parseUsers(RangerService service) { + if (service == null || service.getConfigs() == null) { + return Collections.emptyList(); + } + return parseUsers(service.getConfigs().get(CONFIG_NAME)); + } + + public static List parseUsers(String configValue) { + if (StringUtils.isBlank(configValue)) { + return Collections.emptyList(); + } + return Arrays.stream(configValue.split(",")) + .map(String::trim) + .filter(StringUtils::isNotBlank) + .filter(user -> !"*".equals(user)) + .collect(Collectors.toList()); + } + + public static Map> normalizeServiceAllowedUsers(Map> serviceAllowedUsers) { + if (serviceAllowedUsers == null || serviceAllowedUsers.isEmpty()) { + return Collections.emptyMap(); + } + Map> normalized = new LinkedHashMap<>(); + for (Map.Entry> entry : serviceAllowedUsers.entrySet()) { + if (StringUtils.isBlank(entry.getKey())) { + continue; + } + List users = entry.getValue() == null ? Collections.emptyList() : parseUsers(String.join(",", entry.getValue())); + if (users.isEmpty()) { + continue; + } + normalized.put(entry.getKey().trim(), List.copyOf(users)); + } + return Collections.unmodifiableMap(normalized); + } + + /** Converts plan allow-list to ingestor lookup map; skips repos with no users (same as static site config). */ + public static Map> toAllowedUserSets(Map> serviceAllowedUsers) { + Map> normalized = normalizeServiceAllowedUsers(serviceAllowedUsers); + if (normalized.isEmpty()) { + return Collections.emptyMap(); + } + Map> allowed = new LinkedHashMap<>(); + for (Map.Entry> entry : normalized.entrySet()) { + allowed.put(entry.getKey(), new LinkedHashSet<>(entry.getValue())); + } + return Collections.unmodifiableMap(allowed); + } +} diff --git a/agents-common/src/main/java/org/apache/ranger/audit/partition/exception/PartitionPlanException.java b/agents-common/src/main/java/org/apache/ranger/audit/partition/exception/PartitionPlanException.java new file mode 100644 index 00000000000..92726af30fa --- /dev/null +++ b/agents-common/src/main/java/org/apache/ranger/audit/partition/exception/PartitionPlanException.java @@ -0,0 +1,32 @@ +/* + * 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.partition.exception; + +public class PartitionPlanException extends RuntimeException { + private static final long serialVersionUID = 1L; + + public PartitionPlanException(String message) { + super(message); + } + + public PartitionPlanException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/agents-common/src/main/java/org/apache/ranger/audit/partition/model/BufferEntry.java b/agents-common/src/main/java/org/apache/ranger/audit/partition/model/BufferEntry.java new file mode 100644 index 00000000000..bc90262aaca --- /dev/null +++ b/agents-common/src/main/java/org/apache/ranger/audit/partition/model/BufferEntry.java @@ -0,0 +1,73 @@ +/* + * 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.partition.model; + +import com.fasterxml.jackson.annotation.JsonAutoDetect; +import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility; +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonProperty; + +import java.io.Serializable; +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +@JsonAutoDetect(getterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE, fieldVisibility = Visibility.ANY) +@JsonInclude(JsonInclude.Include.NON_NULL) +public class BufferEntry implements Serializable { + private static final long serialVersionUID = 1L; + + private final List partitions; + + @JsonCreator + public BufferEntry(@JsonProperty("partitions") List partitions) { + if (partitions == null || partitions.isEmpty()) { + this.partitions = Collections.emptyList(); + } else { + this.partitions = List.copyOf(partitions); + } + } + + public static BufferEntry empty() { + return new BufferEntry(Collections.emptyList()); + } + + public List getPartitions() { + return partitions; + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (obj == null || getClass() != obj.getClass()) { + return false; + } + BufferEntry other = (BufferEntry) obj; + return Objects.equals(partitions, other.partitions); + } + + @Override + public int hashCode() { + return Objects.hash(partitions); + } +} diff --git a/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PartitionPlan.java b/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PartitionPlan.java new file mode 100644 index 00000000000..ffcff2ad492 --- /dev/null +++ b/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PartitionPlan.java @@ -0,0 +1,260 @@ +/* + * 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.partition.model; + +import com.fasterxml.jackson.annotation.JsonAutoDetect; +import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility; +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonProperty; +import org.apache.ranger.audit.partition.PartitionPlanValidator; +import org.apache.ranger.audit.partition.exception.PartitionPlanException; +import org.apache.ranger.authorization.utils.JsonUtils; + +import java.io.Serializable; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +@JsonAutoDetect(getterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE, fieldVisibility = Visibility.ANY) +@JsonInclude(JsonInclude.Include.NON_NULL) +public class PartitionPlan implements Serializable { + private static final long serialVersionUID = 1L; + + private final String topic; + private final int version; + private final int topicPartitionCount; + private final String updatedAt; + private final String updatedBy; + private final Map plugins; + private final BufferEntry buffer; + /** Per-repo audit POST allow-list ({@code policy.download.auth.users}); each listed repo needs ≥1 user. */ + private final Map> serviceAllowedUsers; + + @JsonCreator + public PartitionPlan(@JsonProperty("topic") String topic, @JsonProperty("version") int version, @JsonProperty("topicPartitionCount") int topicPartitionCount, @JsonProperty("updatedAt") String updatedAt, @JsonProperty("updatedBy") String updatedBy, @JsonProperty("plugins") Map plugins, @JsonProperty("buffer") BufferEntry buffer, @JsonProperty("serviceAllowedUsers") Map> serviceAllowedUsers) { + this.topic = topic; + this.version = version; + this.topicPartitionCount = topicPartitionCount; + this.updatedAt = updatedAt; + this.updatedBy = updatedBy; + this.plugins = copyPlugins(plugins); + this.buffer = buffer != null ? buffer : BufferEntry.empty(); + this.serviceAllowedUsers = copyServiceAllowedUsers(serviceAllowedUsers); + } + + public String getTopic() { + return topic; + } + + public int getVersion() { + return version; + } + + public int getTopicPartitionCount() { + return topicPartitionCount; + } + + public String getUpdatedAt() { + return updatedAt; + } + + public String getUpdatedBy() { + return updatedBy; + } + + public Map getPlugins() { + return plugins; + } + + public BufferEntry getBuffer() { + return buffer; + } + + public Map> getServiceAllowedUsers() { + return serviceAllowedUsers; + } + + /** Compares routing payload; ignores version, updatedAt, updatedBy, and serviceAllowedUsers. */ + public boolean sameContentAs(PartitionPlan other) { + if (other == null) { + return false; + } + return topicPartitionCount == other.topicPartitionCount + && Objects.equals(topic, other.topic) + && Objects.equals(plugins, other.plugins) + && Objects.equals(buffer, other.buffer); + } + + public Builder toBuilder() { + return new Builder(this); + } + + public static Builder builder() { + return new Builder(); + } + + public String toJson() { + String json = JsonUtils.objectToJson(this); + if (json == null) { + throw new PartitionPlanException("Failed to serialize partition plan"); + } + return json; + } + + public static PartitionPlan fromJson(String json) { + try { + PartitionPlan plan = JsonUtils.jsonToObject(json, PartitionPlan.class); + PartitionPlanValidator.validate(plan); + return plan; + } catch (PartitionPlanException e) { + throw e; + } catch (Exception e) { + throw new PartitionPlanException("Failed to deserialize partition plan", e); + } + } + + @Override + public boolean equals(Object otherPartitionPlanObj) { + if (this == otherPartitionPlanObj) { + return true; + } + if (otherPartitionPlanObj == null || getClass() != otherPartitionPlanObj.getClass()) { + return false; + } + PartitionPlan otherPartitionPlan = (PartitionPlan) otherPartitionPlanObj; + return version == otherPartitionPlan.version + && topicPartitionCount == otherPartitionPlan.topicPartitionCount + && Objects.equals(topic, otherPartitionPlan.topic) + && Objects.equals(updatedAt, otherPartitionPlan.updatedAt) + && Objects.equals(updatedBy, otherPartitionPlan.updatedBy) + && Objects.equals(plugins, otherPartitionPlan.plugins) + && Objects.equals(buffer, otherPartitionPlan.buffer) + && Objects.equals(serviceAllowedUsers, otherPartitionPlan.serviceAllowedUsers); + } + + @Override + public int hashCode() { + return Objects.hash(topic, version, topicPartitionCount, updatedAt, updatedBy, plugins, buffer, serviceAllowedUsers); + } + + @Override + public String toString() { + return "PartitionPlan{topic='" + topic + "', version=" + version + ", topicPartitionCount=" + topicPartitionCount + ", plugins=" + plugins.keySet() + ", bufferSize=" + buffer.getPartitions().size() + ", serviceAllowedUsers=" + serviceAllowedUsers.keySet() + '}'; + } + + private static Map copyPlugins(Map plugins) { + if (plugins == null || plugins.isEmpty()) { + return Collections.emptyMap(); + } + return Collections.unmodifiableMap(new LinkedHashMap<>(plugins)); + } + + private static Map> copyServiceAllowedUsers(Map> serviceAllowedUsers) { + if (serviceAllowedUsers == null || serviceAllowedUsers.isEmpty()) { + return Collections.emptyMap(); + } + Map> copy = new LinkedHashMap<>(); + for (Map.Entry> entry : serviceAllowedUsers.entrySet()) { + if (entry.getKey() == null || entry.getKey().isBlank()) { + continue; + } + List users = entry.getValue() == null ? Collections.emptyList() : List.copyOf(entry.getValue()); + copy.put(entry.getKey().trim(), users); + } + return Collections.unmodifiableMap(copy); + } + + public static final class Builder { + private String topic; + private int version = 1; + private int topicPartitionCount; + private String updatedAt; + private String updatedBy; + private Map plugins = new LinkedHashMap<>(); + private BufferEntry buffer = BufferEntry.empty(); + private Map> serviceAllowedUsers = new LinkedHashMap<>(); + + private Builder() { + } + + private Builder(PartitionPlan plan) { + this.topic = plan.topic; + this.version = plan.version; + this.topicPartitionCount = plan.topicPartitionCount; + this.updatedAt = plan.updatedAt; + this.updatedBy = plan.updatedBy; + this.plugins = new LinkedHashMap<>(plan.plugins); + this.buffer = plan.buffer; + this.serviceAllowedUsers = new LinkedHashMap<>(plan.serviceAllowedUsers); + } + + public Builder topic(String topic) { + this.topic = topic; + return this; + } + + public Builder version(int version) { + this.version = version; + return this; + } + + public Builder topicPartitionCount(int topicPartitionCount) { + this.topicPartitionCount = topicPartitionCount; + return this; + } + + public Builder updatedAt(String updatedAt) { + this.updatedAt = updatedAt; + return this; + } + + public Builder updatedBy(String updatedBy) { + this.updatedBy = updatedBy; + return this; + } + + public Builder plugins(Map plugins) { + this.plugins = plugins == null ? new LinkedHashMap<>() : new LinkedHashMap<>(plugins); + return this; + } + + public Builder putPlugin(String pluginId, PluginEntry entry) { + this.plugins.put(pluginId, entry); + return this; + } + + public Builder buffer(BufferEntry buffer) { + this.buffer = buffer != null ? buffer : BufferEntry.empty(); + return this; + } + + public Builder serviceAllowedUsers(Map> serviceAllowedUsers) { + this.serviceAllowedUsers = serviceAllowedUsers == null ? new LinkedHashMap<>() : new LinkedHashMap<>(serviceAllowedUsers); + return this; + } + + public PartitionPlan build() { + return new PartitionPlan(topic, version, topicPartitionCount, updatedAt, updatedBy, plugins, buffer, serviceAllowedUsers); + } + } +} diff --git a/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PluginEntry.java b/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PluginEntry.java new file mode 100644 index 00000000000..5419ae48310 --- /dev/null +++ b/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PluginEntry.java @@ -0,0 +1,122 @@ +/* + * 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.partition.model; + +import com.fasterxml.jackson.annotation.JsonAutoDetect; +import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility; +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonProperty; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Objects; + +@JsonAutoDetect(getterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE, fieldVisibility = Visibility.ANY) +@JsonInclude(JsonInclude.Include.NON_NULL) +public class PluginEntry implements Serializable { + private static final long serialVersionUID = 1L; + + private final List partitions; + private final List services; + + @JsonCreator + public PluginEntry(@JsonProperty("partitions") List partitions, @JsonProperty("services") List services) { + this.partitions = copyPartitions(partitions); + this.services = copyServices(services); + } + + public static PluginEntry ofPartitions(int... partitionIds) { + List ids = new ArrayList<>(partitionIds.length); + for (int id : partitionIds) { + ids.add(id); + } + return new PluginEntry(ids, Collections.emptyList()); + } + + public static PluginEntry empty() { + return new PluginEntry(Collections.emptyList(), Collections.emptyList()); + } + + public List getPartitions() { + return partitions; + } + + public List getServices() { + return services; + } + + public PluginEntry withPartitions(List newPartitions) { + return new PluginEntry(newPartitions, services); + } + + public PluginEntry withServices(List newServices) { + return new PluginEntry(partitions, newServices); + } + + public PluginEntry addService(String serviceName) { + if (serviceName == null || serviceName.isBlank()) { + return this; + } + LinkedHashSet merged = new LinkedHashSet<>(services); + merged.add(serviceName.trim()); + return new PluginEntry(partitions, List.copyOf(merged)); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (obj == null || getClass() != obj.getClass()) { + return false; + } + PluginEntry other = (PluginEntry) obj; + return Objects.equals(partitions, other.partitions) && Objects.equals(services, other.services); + } + + @Override + public int hashCode() { + return Objects.hash(partitions, services); + } + + private static List copyPartitions(List partitions) { + if (partitions == null || partitions.isEmpty()) { + return Collections.emptyList(); + } + return List.copyOf(partitions); + } + + private static List copyServices(List services) { + if (services == null || services.isEmpty()) { + return Collections.emptyList(); + } + LinkedHashSet unique = new LinkedHashSet<>(); + for (String service : services) { + if (service != null && !service.isBlank()) { + unique.add(service.trim()); + } + } + return List.copyOf(unique); + } +} diff --git a/agents-common/src/test/java/org/apache/ranger/audit/partition/AuditPartitionPlanAdminConfigTest.java b/agents-common/src/test/java/org/apache/ranger/audit/partition/AuditPartitionPlanAdminConfigTest.java new file mode 100644 index 00000000000..8d94f30a913 --- /dev/null +++ b/agents-common/src/test/java/org/apache/ranger/audit/partition/AuditPartitionPlanAdminConfigTest.java @@ -0,0 +1,47 @@ +/* + * 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.partition; + +import org.apache.hadoop.conf.Configuration; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class AuditPartitionPlanAdminConfigTest { + @Test + public void testDefaultPartitionsPerPlugin() { + Configuration config = new Configuration(false); + assertEquals(3, AuditPartitionPlanAdminConfig.resolvePartitionsPerPlugin("hdfs", config)); + } + + @Test + public void testGlobalDefaultFromSiteProperty() { + Configuration config = new Configuration(false); + config.set(AuditPartitionPlanConstants.PROP_ADMIN_PARTITIONS_PER_PLUGIN, "6"); + assertEquals(6, AuditPartitionPlanAdminConfig.resolvePartitionsPerPlugin("hdfs", config)); + } + + @Test + public void testPerPluginOverride() { + Configuration config = new Configuration(false); + config.set(AuditPartitionPlanConstants.PROP_ADMIN_PARTITIONS_PER_PLUGIN, "3"); + config.set(AuditPartitionPlanConstants.PROP_ADMIN_PLUGIN_PARTITION_OVERRIDE_PREFIX + "hiveServer2", "9"); + assertEquals(3, AuditPartitionPlanAdminConfig.resolvePartitionsPerPlugin("hdfs", config)); + assertEquals(9, AuditPartitionPlanAdminConfig.resolvePartitionsPerPlugin("hiveServer2", config)); + } +} diff --git a/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanAllocatorTest.java b/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanAllocatorTest.java new file mode 100644 index 00000000000..7baa39120c3 --- /dev/null +++ b/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanAllocatorTest.java @@ -0,0 +1,164 @@ +/* + * 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.partition; + +import org.apache.ranger.audit.partition.exception.PartitionPlanException; +import org.apache.ranger.audit.partition.model.PartitionPlan; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertIterableEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class PartitionPlanAllocatorTest { + private PartitionPlan initialPlan; + + @BeforeEach + public void setUp() { + initialPlan = PartitionPlanTestSupport.preAssignedPlan(); + } + + @Test + public void testPromotePluginFromBuffer() { + PartitionPlan next = PartitionPlanAllocator.promotePlugin(initialPlan, "trino", 3, "ops"); + + assertEquals(2, next.getVersion()); + assertEquals(9, next.getTopicPartitionCount()); + assertIterableEquals(List.of(7, 8, 9), next.getPlugins().get("trino").getPartitions()); + assertIterableEquals(List.of(), next.getBuffer().getPartitions()); + assertIterableEquals(List.of(1, 2, 3), next.getPlugins().get("hdfs").getPartitions()); + assertIterableEquals(List.of(4, 5, 6), next.getPlugins().get("hiveServer2").getPartitions()); + } + + @Test + public void testPromotePluginGrowsTopicWhenBufferInsufficient() { + PartitionPlan seed = PartitionPlanTestSupport.seedPlan(); + PartitionPlan next = PartitionPlanAllocator.promotePlugin(seed, "trino", 12, "ops"); + + assertEquals(2, next.getVersion()); + assertEquals(12, next.getTopicPartitionCount()); + assertIterableEquals(List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12), next.getPlugins().get("trino").getPartitions()); + assertIterableEquals(List.of(), next.getBuffer().getPartitions()); + } + + @Test + public void testOnboardServicePromotesPluginAndAddsService() { + PartitionPlan seed = PartitionPlanTestSupport.seedPlan(); + PartitionPlan next = PartitionPlanAllocator.onboardService(seed, "hiveServer2", "dev_hive", 6, "admin"); + + assertEquals(2, next.getVersion()); + assertEquals(9, next.getTopicPartitionCount()); + assertIterableEquals(List.of(1, 2, 3, 4, 5, 6), next.getPlugins().get("hiveServer2").getPartitions()); + assertIterableEquals(List.of("dev_hive"), next.getPlugins().get("hiveServer2").getServices()); + assertIterableEquals(List.of(7, 8, 9), next.getBuffer().getPartitions()); + } + + @Test + public void testOnboardServiceAddsToExistingPlugin() { + PartitionPlan promoted = PartitionPlanAllocator.onboardService(PartitionPlanTestSupport.seedPlan(), "hiveServer2", "dev_hive", 6, "admin"); + PartitionPlan next = PartitionPlanAllocator.onboardService(promoted, "hiveServer2", "prod_hive", 6, "admin"); + + assertEquals(3, next.getVersion()); + assertIterableEquals(List.of("dev_hive", "prod_hive"), next.getPlugins().get("hiveServer2").getServices()); + assertIterableEquals(List.of(1, 2, 3, 4, 5, 6), next.getPlugins().get("hiveServer2").getPartitions()); + } + + @Test + public void testRemoveService() { + PartitionPlan onboarded = PartitionPlanAllocator.onboardService(PartitionPlanTestSupport.seedPlan(), "hiveServer2", "dev_hive", 6, "admin"); + PartitionPlan next = PartitionPlanAllocator.removeService(onboarded, "dev_hive", "admin"); + + assertEquals(3, next.getVersion()); + assertIterableEquals(List.of(), next.getPlugins().get("hiveServer2").getServices()); + } + + @Test + public void testScalePluginAppendsTailOnly() { + PartitionPlan promoted = PartitionPlanAllocator.promotePlugin(initialPlan, "trino", 3, "ops"); + PartitionPlan scaled = PartitionPlanAllocator.scalePlugin(promoted, "hiveServer2", 3, "ops"); + + assertEquals(3, scaled.getVersion()); + assertEquals(12, scaled.getTopicPartitionCount()); + assertIterableEquals(List.of(4, 5, 6, 10, 11, 12), scaled.getPlugins().get("hiveServer2").getPartitions()); + assertIterableEquals(List.of(1, 2, 3), scaled.getPlugins().get("hdfs").getPartitions()); + assertIterableEquals(List.of(7, 8, 9), scaled.getPlugins().get("trino").getPartitions()); + } + + @Test + public void testPromoteAlreadyConfiguredPluginFails() { + PartitionPlanException error = assertThrows(PartitionPlanException.class, + () -> PartitionPlanAllocator.promotePlugin(initialPlan, "hdfs", 1, "ops")); + assertTrue(error.getMessage().contains("requested 1")); + } + + @Test + public void testIsPromoteAlreadyAppliedWhenPluginAndCountMatch() { + PartitionPlan promoted = PartitionPlanAllocator.promotePlugin(initialPlan, "trino", 3, "ops"); + + assertTrue(PartitionPlanAllocator.isPromoteAlreadyApplied(promoted, "trino", 3)); + assertFalse(PartitionPlanAllocator.isPromoteAlreadyApplied(promoted, "trino", 5)); + } + + @Test + public void testIsOnboardAlreadyAppliedWhenServiceAndPluginMatch() { + PartitionPlan onboarded = PartitionPlanAllocator.onboardService(PartitionPlanTestSupport.seedPlan(), "hiveServer2", "dev_hive", 6, "admin"); + + assertTrue(PartitionPlanAllocator.isOnboardAlreadyApplied(onboarded, "hiveServer2", "dev_hive", 6)); + assertFalse(PartitionPlanAllocator.isOnboardAlreadyApplied(onboarded, "hiveServer2", "prod_hive", 6)); + } + + @Test + public void testPromoteConflictWhenPartitionCountDiffers() { + PartitionPlan promoted = PartitionPlanAllocator.promotePlugin(initialPlan, "trino", 3, "ops"); + + PartitionPlanException error = assertThrows(PartitionPlanException.class, + () -> PartitionPlanAllocator.promotePlugin(promoted, "trino", 5, "ops")); + + assertTrue(error.getMessage().contains("requested 5")); + } + + @Test + public void testScaleUnknownPluginFails() { + assertThrows(PartitionPlanException.class, () -> PartitionPlanAllocator.scalePlugin(initialPlan, "trino", 2, "ops")); + } + + @Test + public void testUpdateServiceAllowedUsersBumpsVersionWhenMapChanges() { + PartitionPlan next = PartitionPlanAllocator.updateServiceAllowedUsers( + initialPlan, Map.of("dev_hive", List.of("hive")), "admin"); + + assertEquals(2, next.getVersion()); + assertIterableEquals(List.of("hive"), next.getServiceAllowedUsers().get("dev_hive")); + } + + @Test + public void testUpdateServiceAllowedUsersIsNoOpWhenUnchanged() { + PartitionPlan withUsers = PartitionPlanAllocator.updateServiceAllowedUsers( + initialPlan, Map.of("dev_hive", List.of("hive")), "admin"); + PartitionPlan unchanged = PartitionPlanAllocator.updateServiceAllowedUsers( + withUsers, Map.of("dev_hive", List.of("hive")), "admin"); + + assertEquals(withUsers, unchanged); + } +} diff --git a/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanRoutingUtilsTest.java b/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanRoutingUtilsTest.java new file mode 100644 index 00000000000..edce4104cfa --- /dev/null +++ b/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanRoutingUtilsTest.java @@ -0,0 +1,43 @@ +/* + * 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.partition; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +public class PartitionPlanRoutingUtilsTest { + @Test + public void testToKafkaPartitionIndexUsesZeroBasedMapping() { + assertEquals(0, PartitionPlanRoutingUtils.toKafkaPartitionIndex(1)); + assertEquals(8, PartitionPlanRoutingUtils.toKafkaPartitionIndex(9)); + } + + @Test + public void testResolveKafkaPartitionIndexClampsToTopicSize() { + assertEquals(8, PartitionPlanRoutingUtils.resolveKafkaPartitionIndex(9, 9)); + assertEquals(8, PartitionPlanRoutingUtils.resolveKafkaPartitionIndex(12, 9)); + } + + @Test + public void testHashToSlotIndexUsesFloorModForMinHashCode() { + String minHashKey = "polygenelubricants"; + assertEquals(Integer.MIN_VALUE, minHashKey.hashCode()); + assertEquals(2, PartitionPlanRoutingUtils.hashToSlotIndex(minHashKey, 5)); + } +} diff --git a/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanTestSupport.java b/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanTestSupport.java new file mode 100644 index 00000000000..836f48f0083 --- /dev/null +++ b/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanTestSupport.java @@ -0,0 +1,62 @@ +/* + * 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.partition; + +import org.apache.ranger.audit.partition.model.BufferEntry; +import org.apache.ranger.audit.partition.model.PartitionPlan; +import org.apache.ranger.audit.partition.model.PluginEntry; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +final class PartitionPlanTestSupport { + private PartitionPlanTestSupport() { + } + + static PartitionPlan seedPlan() { + return PartitionPlan.builder() + .topic(AuditPartitionPlanConstants.DEFAULT_AUDIT_TOPIC) + .version(AuditPartitionPlanConstants.INITIAL_PLAN_VERSION) + .topicPartitionCount(9) + .buffer(new BufferEntry(partitionRange(1, 9))) + .build(); + } + + static PartitionPlan preAssignedPlan() { + Map plugins = new LinkedHashMap<>(); + plugins.put("hdfs", PluginEntry.ofPartitions(1, 2, 3)); + plugins.put("hiveServer2", PluginEntry.ofPartitions(4, 5, 6)); + return PartitionPlan.builder() + .topic(AuditPartitionPlanConstants.DEFAULT_AUDIT_TOPIC) + .version(AuditPartitionPlanConstants.INITIAL_PLAN_VERSION) + .topicPartitionCount(9) + .plugins(plugins) + .buffer(new BufferEntry(partitionRange(7, 9))) + .build(); + } + + private static List partitionRange(int startInclusive, int endInclusive) { + List ids = new ArrayList<>(); + for (int id = startInclusive; id <= endInclusive; id++) { + ids.add(id); + } + return ids; + } +} diff --git a/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanValidatorTest.java b/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanValidatorTest.java new file mode 100644 index 00000000000..f39e462adc8 --- /dev/null +++ b/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanValidatorTest.java @@ -0,0 +1,154 @@ +/* + * 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.partition; + +import org.apache.ranger.audit.partition.exception.PartitionPlanException; +import org.apache.ranger.audit.partition.model.BufferEntry; +import org.apache.ranger.audit.partition.model.PartitionPlan; +import org.apache.ranger.audit.partition.model.PluginEntry; +import org.junit.jupiter.api.Test; + +import java.util.LinkedHashMap; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertThrows; + +public class PartitionPlanValidatorTest { + @Test + public void testValidateAcceptsSeedPlan() { + PartitionPlan plan = PartitionPlanTestSupport.seedPlan(); + assertDoesNotThrow(() -> PartitionPlanValidator.validate(plan, 9)); + } + + @Test + public void testValidateAcceptsPreAssignedPlan() { + PartitionPlan plan = PartitionPlanTestSupport.preAssignedPlan(); + assertDoesNotThrow(() -> PartitionPlanValidator.validate(plan)); + } + + @Test + public void testValidateRejectsDuplicatePartitionIds() { + Map plugins = new LinkedHashMap<>(); + plugins.put("hdfs", PluginEntry.ofPartitions(1, 2)); + PartitionPlan plan = PartitionPlan.builder() + .topic("ranger_audits") + .version(1) + .topicPartitionCount(3) + .plugins(plugins) + .buffer(new BufferEntry(java.util.List.of(2, 3))) + .build(); + assertThrows(PartitionPlanException.class, () -> PartitionPlanValidator.validate(plan)); + } + + @Test + public void testValidateRejectsUnionSizeMismatch() { + PartitionPlan plan = PartitionPlan.builder() + .topic("ranger_audits") + .version(1) + .topicPartitionCount(10) + .buffer(new BufferEntry(java.util.List.of(1, 2, 3, 4, 5, 6, 7, 8, 9))) + .build(); + assertThrows(PartitionPlanException.class, () -> PartitionPlanValidator.validate(plan)); + } + + @Test + public void testValidateRejectsZeroBasedPartitionId() { + PartitionPlan plan = PartitionPlan.builder() + .topic("ranger_audits") + .version(1) + .topicPartitionCount(1) + .buffer(new BufferEntry(java.util.List.of(0))) + .build(); + assertThrows(PartitionPlanException.class, () -> PartitionPlanValidator.validate(plan)); + } + + @Test + public void testValidateRejectsDuplicateServiceAssignment() { + Map plugins = new LinkedHashMap<>(); + plugins.put("hdfs", new PluginEntry(java.util.List.of(1, 2, 3), java.util.List.of("dev_hdfs"))); + plugins.put("hiveServer2", new PluginEntry(java.util.List.of(4, 5, 6), java.util.List.of("dev_hdfs"))); + PartitionPlan plan = PartitionPlan.builder() + .topic("ranger_audits") + .version(1) + .topicPartitionCount(9) + .plugins(plugins) + .buffer(new BufferEntry(java.util.List.of(7, 8, 9))) + .build(); + assertThrows(PartitionPlanException.class, () -> PartitionPlanValidator.validate(plan)); + } + + @Test + public void testValidateRejectsKafkaPartitionCountBelowPlan() { + PartitionPlan plan = PartitionPlanTestSupport.seedPlan(); + assertThrows(PartitionPlanException.class, () -> PartitionPlanValidator.validate(plan, 5)); + } + + @Test + public void testValidateAcceptsKafkaPartitionCountAbovePlan() { + PartitionPlan plan = PartitionPlanTestSupport.seedPlan(); + assertDoesNotThrow(() -> PartitionPlanValidator.validate(plan, 30)); + } + + @Test + public void testValidateAppendOnlyRejectsReshuffle() { + PartitionPlan current = PartitionPlanTestSupport.preAssignedPlan(); + Map reshuffled = new LinkedHashMap<>(); + reshuffled.put("hdfs", PluginEntry.ofPartitions(1, 2, 3, 4)); + reshuffled.put("hiveServer2", PluginEntry.ofPartitions(5, 6)); + PartitionPlan proposed = PartitionPlan.builder() + .topic("ranger_audits") + .version(2) + .topicPartitionCount(9) + .plugins(reshuffled) + .buffer(new BufferEntry(java.util.List.of(7, 8, 9))) + .build(); + assertThrows(PartitionPlanException.class, () -> PartitionPlanValidator.validateAppendOnly(current, proposed)); + } + + @Test + public void testValidateAppendOnlyAcceptsTailGrowth() { + PartitionPlan current = PartitionPlanTestSupport.preAssignedPlan(); + Map grown = new LinkedHashMap<>(); + grown.put("hdfs", PluginEntry.ofPartitions(1, 2, 3)); + grown.put("hiveServer2", PluginEntry.ofPartitions(4, 5, 6, 10, 11, 12)); + PartitionPlan proposed = PartitionPlan.builder() + .topic("ranger_audits") + .version(2) + .topicPartitionCount(12) + .plugins(grown) + .buffer(new BufferEntry(java.util.List.of(7, 8, 9))) + .build(); + assertDoesNotThrow(() -> PartitionPlanValidator.validateAppendOnly(current, proposed)); + } + + @Test + public void testValidateRejectsEmptyServiceAllowedUsers() { + Map> allowlists = new LinkedHashMap<>(); + allowlists.put("dev_hive", java.util.Collections.emptyList()); + PartitionPlan plan = PartitionPlan.builder() + .topic("ranger_audits") + .version(1) + .topicPartitionCount(9) + .plugins(PartitionPlanTestSupport.preAssignedPlan().getPlugins()) + .buffer(new BufferEntry(java.util.List.of(7, 8, 9))) + .serviceAllowedUsers(allowlists) + .build(); + assertThrows(PartitionPlanException.class, () -> PartitionPlanValidator.validate(plan)); + } +} diff --git a/agents-common/src/test/java/org/apache/ranger/audit/partition/PolicyDownloadAuthUsersUtilTest.java b/agents-common/src/test/java/org/apache/ranger/audit/partition/PolicyDownloadAuthUsersUtilTest.java new file mode 100644 index 00000000000..649303446a8 --- /dev/null +++ b/agents-common/src/test/java/org/apache/ranger/audit/partition/PolicyDownloadAuthUsersUtilTest.java @@ -0,0 +1,75 @@ +/* + * 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.partition; + +import org.apache.ranger.plugin.model.RangerService; +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertIterableEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class PolicyDownloadAuthUsersUtilTest { + @Test + public void testParseUsersFromConfigValue() { + assertIterableEquals(List.of("hive", "hive2"), PolicyDownloadAuthUsersUtil.parseUsers(" hive, hive2 ")); + } + + @Test + public void testParseUsersIgnoresWildcard() { + assertTrue(PolicyDownloadAuthUsersUtil.parseUsers("*").isEmpty()); + } + + @Test + public void testParseUsersFromService() { + RangerService service = new RangerService(); + Map configs = new HashMap<>(); + configs.put(PolicyDownloadAuthUsersUtil.CONFIG_NAME, "hdfs"); + service.setConfigs(configs); + + assertIterableEquals(List.of("hdfs"), PolicyDownloadAuthUsersUtil.parseUsers(service)); + } + + @Test + public void testNormalizeServiceAllowedUsersSkipsEmptyEntries() { + Map> input = new LinkedHashMap<>(); + input.put("dev_hive", List.of("hive")); + input.put("dev_empty", List.of()); + input.put("dev_wildcard", List.of("*")); + + Map> normalized = PolicyDownloadAuthUsersUtil.normalizeServiceAllowedUsers(input); + + assertEquals(1, normalized.size()); + assertIterableEquals(List.of("hive"), normalized.get("dev_hive")); + } + + @Test + public void testToAllowedUserSets() { + Map> input = Map.of("dev_hive", List.of("hive", "hive2")); + + Map> allowed = PolicyDownloadAuthUsersUtil.toAllowedUserSets(input); + + assertEquals(Set.of("hive", "hive2"), allowed.get("dev_hive")); + } +} diff --git a/agents-common/src/test/java/org/apache/ranger/audit/partition/model/PartitionPlanJsonTest.java b/agents-common/src/test/java/org/apache/ranger/audit/partition/model/PartitionPlanJsonTest.java new file mode 100644 index 00000000000..fdd03058dca --- /dev/null +++ b/agents-common/src/test/java/org/apache/ranger/audit/partition/model/PartitionPlanJsonTest.java @@ -0,0 +1,73 @@ +/* + * 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.partition.model; + +import org.apache.ranger.audit.partition.AuditPartitionPlanConstants; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertIterableEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +public class PartitionPlanJsonTest { + @Test + public void testRoundTripSeedJson() { + String seedJson = "{\"version\":1,\"topic\":\"ranger_audits\",\"topicPartitionCount\":9,\"plugins\":{},\"buffer\":{\"partitions\":[1,2,3,4,5,6,7,8,9]}}"; + PartitionPlan plan = PartitionPlan.fromJson(seedJson); + + assertEquals(AuditPartitionPlanConstants.DEFAULT_AUDIT_TOPIC, plan.getTopic()); + assertEquals(1, plan.getVersion()); + assertEquals(9, plan.getTopicPartitionCount()); + assertEquals(0, plan.getPlugins().size()); + assertEquals(9, plan.getBuffer().getPartitions().size()); + + String roundTrip = plan.toJson(); + assertNotNull(roundTrip); + PartitionPlan parsedAgain = PartitionPlan.fromJson(roundTrip); + assertEquals(plan, parsedAgain); + } + + @Test + public void testRoundTripOnboardedPluginJson() { + String json = "{\"version\":2,\"topic\":\"ranger_audits\",\"topicPartitionCount\":9," + + "\"plugins\":{\"hiveServer2\":{\"partitions\":[1,2,3,4,5,6],\"services\":[\"dev_hive\",\"prod_hive\"]}}," + + "\"buffer\":{\"partitions\":[7,8,9]}}"; + PartitionPlan plan = PartitionPlan.fromJson(json); + + assertEquals(2, plan.getVersion()); + assertEquals(2, plan.getPlugins().get("hiveServer2").getServices().size()); + assertEquals(6, plan.getPlugins().get("hiveServer2").getPartitions().size()); + } + + @Test + public void testRoundTripServiceAllowedUsersJson() { + String json = "{\"version\":3,\"topic\":\"ranger_audits\",\"topicPartitionCount\":9," + + "\"plugins\":{},\"buffer\":{\"partitions\":[1,2,3,4,5,6,7,8,9]}," + + "\"serviceAllowedUsers\":{\"dev_hive\":[\"hive\"],\"dev_ozone\":[\"om\"]}}"; + PartitionPlan plan = PartitionPlan.fromJson(json); + + assertEquals(3, plan.getVersion()); + assertIterableEquals(List.of("hive"), plan.getServiceAllowedUsers().get("dev_hive")); + assertIterableEquals(List.of("om"), plan.getServiceAllowedUsers().get("dev_ozone")); + + PartitionPlan parsedAgain = PartitionPlan.fromJson(plan.toJson()); + assertEquals(plan, parsedAgain); + } +} From 9f6bad1249a5ef77b02f5d42926edd827dd02e5e Mon Sep 17 00:00:00 2001 From: ramk Date: Thu, 6 Aug 2026 13:02:51 +0530 Subject: [PATCH 2/4] RANGER-5719: Remove unused partition-plan library APIs Drop scalePlugin, replacePlan, isPromoteAlreadyApplied, sameContentAs, toAllowedUserSets, hashToSlotIndex, and related helpers that no production caller uses in the 5721/5722 design; can reintroduce when scaling or bulk replace is wired. --- .../partition/PartitionPlanAllocator.java | 36 ------------------- .../partition/PartitionPlanRoutingUtils.java | 16 --------- .../PolicyDownloadAuthUsersUtil.java | 22 ------------ .../audit/partition/model/PartitionPlan.java | 11 ------ .../audit/partition/model/PluginEntry.java | 4 --- .../partition/PartitionPlanAllocatorTest.java | 25 ------------- .../PartitionPlanRoutingUtilsTest.java | 7 ---- .../PolicyDownloadAuthUsersUtilTest.java | 10 ------ 8 files changed, 131 deletions(-) diff --git a/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanAllocator.java b/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanAllocator.java index a8507d2e115..599ccea921c 100644 --- a/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanAllocator.java +++ b/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanAllocator.java @@ -137,20 +137,6 @@ public static PartitionPlan promotePlugin(PartitionPlan current, String pluginId return commitPlanUpdate(current, updatedBy, topicPartitionCount, plugins, remainingBuffer); } - /** Add more partitions to an existing plugin by appending new tail IDs only. */ - public static PartitionPlan scalePlugin(PartitionPlan current, String pluginId, int additionalPartitions, String updatedBy) { - requireMutationInputs(current, pluginId, additionalPartitions, updatedBy); - if (!current.getPlugins().containsKey(pluginId)) { - throw new PartitionPlanException("Plugin '" + pluginId + "' is not configured; promote it first"); - } - - List pluginIds = new ArrayList<>(current.getPlugins().get(pluginId).getPartitions()); - int topicPartitionCount = appendTailPartitions(pluginIds, current.getTopicPartitionCount(), additionalPartitions, collectAssignedPartitionIds(current)); - - Map plugins = addPluginAssignment(current, pluginId, pluginIds, current.getPlugins().get(pluginId).getServices()); - return commitPlanUpdate(current, updatedBy, topicPartitionCount, plugins, current.getBuffer().getPartitions()); - } - public static boolean isOnboardAlreadyApplied(PartitionPlan current, String pluginId, String serviceName, int partitionCount) { if (current == null || StringUtils.isBlank(serviceName)) { return false; @@ -162,28 +148,6 @@ public static boolean isOnboardAlreadyApplied(PartitionPlan current, String plug return existing.getPartitions().size() == partitionCount && existing.getServices().contains(serviceName.trim()); } - public static boolean isPromoteAlreadyApplied(PartitionPlan current, String pluginId, int partitionCount) { - if (current == null) { - return false; - } - PluginEntry existing = current.getPlugins().get(pluginId); - return existing != null && existing.getPartitions().size() == partitionCount; - } - - /** Applies a merged plan with append-only checks against the current plan. */ - public static PartitionPlan replacePlan(PartitionPlan current, PartitionPlan proposed) { - if (current == null || proposed == null) { - throw new PartitionPlanException("Current and proposed plans are required"); - } - if (!StringUtils.equals(current.getTopic(), proposed.getTopic())) { - throw new PartitionPlanException("Proposed topic must match current topic"); - } - PartitionPlan next = proposed.toBuilder().version(current.getVersion() + 1).build(); - PartitionPlanValidator.validate(next); - PartitionPlanValidator.validateAppendOnly(current, next); - return next; - } - /** Updates audit POST allow-list metadata; bumps version only when the map changes. */ public static PartitionPlan updateServiceAllowedUsers(PartitionPlan current, Map> serviceAllowedUsers, String updatedBy) { if (current == null) { diff --git a/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanRoutingUtils.java b/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanRoutingUtils.java index 65849f756b5..67c1ca6bee5 100644 --- a/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanRoutingUtils.java +++ b/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanRoutingUtils.java @@ -19,8 +19,6 @@ package org.apache.ranger.audit.partition; -import org.apache.commons.lang3.StringUtils; - /** Converts Admin-managed plan partition ids to Kafka producer partition indices. */ public final class PartitionPlanRoutingUtils { private PartitionPlanRoutingUtils() { @@ -37,20 +35,6 @@ public static int toKafkaPartitionIndex(int plannedPartitionId) { return plannedPartitionId - 1; } - /** - * Returns a non-negative slot index in {@code [0, slotCount)} for hash-based buffer routing. - * Uses {@link Math#floorMod(int, int)} so {@code Integer.MIN_VALUE} hash codes are safe. - */ - public static int hashToSlotIndex(String key, int slotCount) { - if (slotCount <= 0) { - return 0; - } - if (StringUtils.isBlank(key)) { - return 0; - } - return Math.floorMod(key.hashCode(), slotCount); - } - /** * Returns the Kafka partition index for a planned id, clamped to the effective topic size when metadata lags. */ diff --git a/agents-common/src/main/java/org/apache/ranger/audit/partition/PolicyDownloadAuthUsersUtil.java b/agents-common/src/main/java/org/apache/ranger/audit/partition/PolicyDownloadAuthUsersUtil.java index 08d6844f288..550eecceafc 100644 --- a/agents-common/src/main/java/org/apache/ranger/audit/partition/PolicyDownloadAuthUsersUtil.java +++ b/agents-common/src/main/java/org/apache/ranger/audit/partition/PolicyDownloadAuthUsersUtil.java @@ -25,23 +25,14 @@ import java.util.Arrays; import java.util.Collections; import java.util.LinkedHashMap; -import java.util.LinkedHashSet; import java.util.List; import java.util.Map; -import java.util.Set; import java.util.stream.Collectors; /** Parses {@code policy.download.auth.users} service config for audit ingestor authorization. */ public final class PolicyDownloadAuthUsersUtil { public static final String CONFIG_NAME = "policy.download.auth.users"; - /** - * Ingestor site key pattern: {@code ranger.audit.ingestor.service..allowed.users}. - * Values originate from Admin {@link #CONFIG_NAME}; partition plan {@code serviceAllowedUsers} - * keys use the same {@code } names (Policy Manager service name). - */ - public static final String INGESTOR_ALLOWED_USERS_SUFFIX = "allowed.users"; - private PolicyDownloadAuthUsersUtil() { } @@ -80,17 +71,4 @@ public static Map> normalizeServiceAllowedUsers(Map> toAllowedUserSets(Map> serviceAllowedUsers) { - Map> normalized = normalizeServiceAllowedUsers(serviceAllowedUsers); - if (normalized.isEmpty()) { - return Collections.emptyMap(); - } - Map> allowed = new LinkedHashMap<>(); - for (Map.Entry> entry : normalized.entrySet()) { - allowed.put(entry.getKey(), new LinkedHashSet<>(entry.getValue())); - } - return Collections.unmodifiableMap(allowed); - } } diff --git a/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PartitionPlan.java b/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PartitionPlan.java index ffcff2ad492..968d449b773 100644 --- a/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PartitionPlan.java +++ b/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PartitionPlan.java @@ -94,17 +94,6 @@ public Map> getServiceAllowedUsers() { return serviceAllowedUsers; } - /** Compares routing payload; ignores version, updatedAt, updatedBy, and serviceAllowedUsers. */ - public boolean sameContentAs(PartitionPlan other) { - if (other == null) { - return false; - } - return topicPartitionCount == other.topicPartitionCount - && Objects.equals(topic, other.topic) - && Objects.equals(plugins, other.plugins) - && Objects.equals(buffer, other.buffer); - } - public Builder toBuilder() { return new Builder(this); } diff --git a/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PluginEntry.java b/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PluginEntry.java index 5419ae48310..876e308d3e2 100644 --- a/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PluginEntry.java +++ b/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PluginEntry.java @@ -66,10 +66,6 @@ public List getServices() { return services; } - public PluginEntry withPartitions(List newPartitions) { - return new PluginEntry(newPartitions, services); - } - public PluginEntry withServices(List newServices) { return new PluginEntry(partitions, newServices); } diff --git a/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanAllocatorTest.java b/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanAllocatorTest.java index 7baa39120c3..e8dd7c9fbc2 100644 --- a/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanAllocatorTest.java +++ b/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanAllocatorTest.java @@ -93,18 +93,6 @@ public void testRemoveService() { assertIterableEquals(List.of(), next.getPlugins().get("hiveServer2").getServices()); } - @Test - public void testScalePluginAppendsTailOnly() { - PartitionPlan promoted = PartitionPlanAllocator.promotePlugin(initialPlan, "trino", 3, "ops"); - PartitionPlan scaled = PartitionPlanAllocator.scalePlugin(promoted, "hiveServer2", 3, "ops"); - - assertEquals(3, scaled.getVersion()); - assertEquals(12, scaled.getTopicPartitionCount()); - assertIterableEquals(List.of(4, 5, 6, 10, 11, 12), scaled.getPlugins().get("hiveServer2").getPartitions()); - assertIterableEquals(List.of(1, 2, 3), scaled.getPlugins().get("hdfs").getPartitions()); - assertIterableEquals(List.of(7, 8, 9), scaled.getPlugins().get("trino").getPartitions()); - } - @Test public void testPromoteAlreadyConfiguredPluginFails() { PartitionPlanException error = assertThrows(PartitionPlanException.class, @@ -112,14 +100,6 @@ public void testPromoteAlreadyConfiguredPluginFails() { assertTrue(error.getMessage().contains("requested 1")); } - @Test - public void testIsPromoteAlreadyAppliedWhenPluginAndCountMatch() { - PartitionPlan promoted = PartitionPlanAllocator.promotePlugin(initialPlan, "trino", 3, "ops"); - - assertTrue(PartitionPlanAllocator.isPromoteAlreadyApplied(promoted, "trino", 3)); - assertFalse(PartitionPlanAllocator.isPromoteAlreadyApplied(promoted, "trino", 5)); - } - @Test public void testIsOnboardAlreadyAppliedWhenServiceAndPluginMatch() { PartitionPlan onboarded = PartitionPlanAllocator.onboardService(PartitionPlanTestSupport.seedPlan(), "hiveServer2", "dev_hive", 6, "admin"); @@ -138,11 +118,6 @@ public void testPromoteConflictWhenPartitionCountDiffers() { assertTrue(error.getMessage().contains("requested 5")); } - @Test - public void testScaleUnknownPluginFails() { - assertThrows(PartitionPlanException.class, () -> PartitionPlanAllocator.scalePlugin(initialPlan, "trino", 2, "ops")); - } - @Test public void testUpdateServiceAllowedUsersBumpsVersionWhenMapChanges() { PartitionPlan next = PartitionPlanAllocator.updateServiceAllowedUsers( diff --git a/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanRoutingUtilsTest.java b/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanRoutingUtilsTest.java index edce4104cfa..6c1de132fc3 100644 --- a/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanRoutingUtilsTest.java +++ b/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanRoutingUtilsTest.java @@ -33,11 +33,4 @@ public void testResolveKafkaPartitionIndexClampsToTopicSize() { assertEquals(8, PartitionPlanRoutingUtils.resolveKafkaPartitionIndex(9, 9)); assertEquals(8, PartitionPlanRoutingUtils.resolveKafkaPartitionIndex(12, 9)); } - - @Test - public void testHashToSlotIndexUsesFloorModForMinHashCode() { - String minHashKey = "polygenelubricants"; - assertEquals(Integer.MIN_VALUE, minHashKey.hashCode()); - assertEquals(2, PartitionPlanRoutingUtils.hashToSlotIndex(minHashKey, 5)); - } } diff --git a/agents-common/src/test/java/org/apache/ranger/audit/partition/PolicyDownloadAuthUsersUtilTest.java b/agents-common/src/test/java/org/apache/ranger/audit/partition/PolicyDownloadAuthUsersUtilTest.java index 649303446a8..51e8cb8853c 100644 --- a/agents-common/src/test/java/org/apache/ranger/audit/partition/PolicyDownloadAuthUsersUtilTest.java +++ b/agents-common/src/test/java/org/apache/ranger/audit/partition/PolicyDownloadAuthUsersUtilTest.java @@ -24,7 +24,6 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import java.util.Set; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertIterableEquals; @@ -63,13 +62,4 @@ public void testNormalizeServiceAllowedUsersSkipsEmptyEntries() { assertEquals(1, normalized.size()); assertIterableEquals(List.of("hive"), normalized.get("dev_hive")); } - - @Test - public void testToAllowedUserSets() { - Map> input = Map.of("dev_hive", List.of("hive", "hive2")); - - Map> allowed = PolicyDownloadAuthUsersUtil.toAllowedUserSets(input); - - assertEquals(Set.of("hive", "hive2"), allowed.get("dev_hive")); - } } From 645fa69a5e9b6fc58afd0b1fa657118f3e38cdd1 Mon Sep 17 00:00:00 2001 From: ramk Date: Thu, 6 Aug 2026 13:08:46 +0530 Subject: [PATCH 3/4] RANGER-5719: Trim remaining unused partition-plan helpers Remove PluginEntry.empty(), the four-arg promotePlugin overload, and DEFAULT_AUDIT_TOPIC (tests use the literal topic name). --- .../audit/partition/AuditPartitionPlanConstants.java | 3 +-- .../ranger/audit/partition/PartitionPlanAllocator.java | 4 ---- .../ranger/audit/partition/model/PluginEntry.java | 4 ---- .../audit/partition/PartitionPlanAllocatorTest.java | 10 +++++----- .../audit/partition/PartitionPlanTestSupport.java | 4 ++-- .../audit/partition/model/PartitionPlanJsonTest.java | 3 +-- 6 files changed, 9 insertions(+), 19 deletions(-) diff --git a/agents-common/src/main/java/org/apache/ranger/audit/partition/AuditPartitionPlanConstants.java b/agents-common/src/main/java/org/apache/ranger/audit/partition/AuditPartitionPlanConstants.java index 38c20c6547f..1c8352893ef 100644 --- a/agents-common/src/main/java/org/apache/ranger/audit/partition/AuditPartitionPlanConstants.java +++ b/agents-common/src/main/java/org/apache/ranger/audit/partition/AuditPartitionPlanConstants.java @@ -20,8 +20,7 @@ package org.apache.ranger.audit.partition; public final class AuditPartitionPlanConstants { - public static final int INITIAL_PLAN_VERSION = 1; - public static final String DEFAULT_AUDIT_TOPIC = "ranger_audits"; + public static final int INITIAL_PLAN_VERSION = 1; /** Default partition slots allocated when a plugin is first promoted from buffer. */ public static final int DEFAULT_PARTITIONS_PER_PLUGIN = 3; diff --git a/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanAllocator.java b/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanAllocator.java index 599ccea921c..b9c92138c08 100644 --- a/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanAllocator.java +++ b/agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanAllocator.java @@ -106,10 +106,6 @@ public static PartitionPlan removeService(PartitionPlan current, String serviceN return commitPlanUpdate(current, updatedBy, current.getTopicPartitionCount(), plugins, current.getBuffer().getPartitions()); } - public static PartitionPlan promotePlugin(PartitionPlan current, String pluginId, int partitionCount, String updatedBy) { - return promotePlugin(current, pluginId, partitionCount, updatedBy, null); - } - /** * Give a plugin its own partitions. Uses buffer IDs first; adds new tail IDs when buffer is too small. * Optionally attaches {@code serviceName} to the new plugin entry. diff --git a/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PluginEntry.java b/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PluginEntry.java index 876e308d3e2..44f2899f7c2 100644 --- a/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PluginEntry.java +++ b/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PluginEntry.java @@ -54,10 +54,6 @@ public static PluginEntry ofPartitions(int... partitionIds) { return new PluginEntry(ids, Collections.emptyList()); } - public static PluginEntry empty() { - return new PluginEntry(Collections.emptyList(), Collections.emptyList()); - } - public List getPartitions() { return partitions; } diff --git a/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanAllocatorTest.java b/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanAllocatorTest.java index e8dd7c9fbc2..09915ead48c 100644 --- a/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanAllocatorTest.java +++ b/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanAllocatorTest.java @@ -41,7 +41,7 @@ public void setUp() { @Test public void testPromotePluginFromBuffer() { - PartitionPlan next = PartitionPlanAllocator.promotePlugin(initialPlan, "trino", 3, "ops"); + PartitionPlan next = PartitionPlanAllocator.promotePlugin(initialPlan, "trino", 3, "ops", null); assertEquals(2, next.getVersion()); assertEquals(9, next.getTopicPartitionCount()); @@ -54,7 +54,7 @@ public void testPromotePluginFromBuffer() { @Test public void testPromotePluginGrowsTopicWhenBufferInsufficient() { PartitionPlan seed = PartitionPlanTestSupport.seedPlan(); - PartitionPlan next = PartitionPlanAllocator.promotePlugin(seed, "trino", 12, "ops"); + PartitionPlan next = PartitionPlanAllocator.promotePlugin(seed, "trino", 12, "ops", null); assertEquals(2, next.getVersion()); assertEquals(12, next.getTopicPartitionCount()); @@ -96,7 +96,7 @@ public void testRemoveService() { @Test public void testPromoteAlreadyConfiguredPluginFails() { PartitionPlanException error = assertThrows(PartitionPlanException.class, - () -> PartitionPlanAllocator.promotePlugin(initialPlan, "hdfs", 1, "ops")); + () -> PartitionPlanAllocator.promotePlugin(initialPlan, "hdfs", 1, "ops", null)); assertTrue(error.getMessage().contains("requested 1")); } @@ -110,10 +110,10 @@ public void testIsOnboardAlreadyAppliedWhenServiceAndPluginMatch() { @Test public void testPromoteConflictWhenPartitionCountDiffers() { - PartitionPlan promoted = PartitionPlanAllocator.promotePlugin(initialPlan, "trino", 3, "ops"); + PartitionPlan promoted = PartitionPlanAllocator.promotePlugin(initialPlan, "trino", 3, "ops", null); PartitionPlanException error = assertThrows(PartitionPlanException.class, - () -> PartitionPlanAllocator.promotePlugin(promoted, "trino", 5, "ops")); + () -> PartitionPlanAllocator.promotePlugin(promoted, "trino", 5, "ops", null)); assertTrue(error.getMessage().contains("requested 5")); } diff --git a/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanTestSupport.java b/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanTestSupport.java index 836f48f0083..ea7074e85d2 100644 --- a/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanTestSupport.java +++ b/agents-common/src/test/java/org/apache/ranger/audit/partition/PartitionPlanTestSupport.java @@ -32,7 +32,7 @@ private PartitionPlanTestSupport() { static PartitionPlan seedPlan() { return PartitionPlan.builder() - .topic(AuditPartitionPlanConstants.DEFAULT_AUDIT_TOPIC) + .topic("ranger_audits") .version(AuditPartitionPlanConstants.INITIAL_PLAN_VERSION) .topicPartitionCount(9) .buffer(new BufferEntry(partitionRange(1, 9))) @@ -44,7 +44,7 @@ static PartitionPlan preAssignedPlan() { plugins.put("hdfs", PluginEntry.ofPartitions(1, 2, 3)); plugins.put("hiveServer2", PluginEntry.ofPartitions(4, 5, 6)); return PartitionPlan.builder() - .topic(AuditPartitionPlanConstants.DEFAULT_AUDIT_TOPIC) + .topic("ranger_audits") .version(AuditPartitionPlanConstants.INITIAL_PLAN_VERSION) .topicPartitionCount(9) .plugins(plugins) diff --git a/agents-common/src/test/java/org/apache/ranger/audit/partition/model/PartitionPlanJsonTest.java b/agents-common/src/test/java/org/apache/ranger/audit/partition/model/PartitionPlanJsonTest.java index fdd03058dca..d10d50c6b92 100644 --- a/agents-common/src/test/java/org/apache/ranger/audit/partition/model/PartitionPlanJsonTest.java +++ b/agents-common/src/test/java/org/apache/ranger/audit/partition/model/PartitionPlanJsonTest.java @@ -17,7 +17,6 @@ package org.apache.ranger.audit.partition.model; -import org.apache.ranger.audit.partition.AuditPartitionPlanConstants; import org.junit.jupiter.api.Test; import java.util.List; @@ -32,7 +31,7 @@ public void testRoundTripSeedJson() { String seedJson = "{\"version\":1,\"topic\":\"ranger_audits\",\"topicPartitionCount\":9,\"plugins\":{},\"buffer\":{\"partitions\":[1,2,3,4,5,6,7,8,9]}}"; PartitionPlan plan = PartitionPlan.fromJson(seedJson); - assertEquals(AuditPartitionPlanConstants.DEFAULT_AUDIT_TOPIC, plan.getTopic()); + assertEquals("ranger_audits", plan.getTopic()); assertEquals(1, plan.getVersion()); assertEquals(9, plan.getTopicPartitionCount()); assertEquals(0, plan.getPlugins().size()); From 37348ab19a8a19304eee42982c99d20aa0e7e99e Mon Sep 17 00:00:00 2001 From: ramk Date: Thu, 6 Aug 2026 14:01:57 +0530 Subject: [PATCH 4/4] RANGER-5719: Remove unused PartitionPlan.Builder.putPlugin No callers in #1137, rangerRelease 5721/5722, or tests; plans are built via plugins(map). --- .../apache/ranger/audit/partition/model/PartitionPlan.java | 5 ----- 1 file changed, 5 deletions(-) diff --git a/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PartitionPlan.java b/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PartitionPlan.java index 968d449b773..d2120a5b5ac 100644 --- a/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PartitionPlan.java +++ b/agents-common/src/main/java/org/apache/ranger/audit/partition/model/PartitionPlan.java @@ -227,11 +227,6 @@ public Builder plugins(Map plugins) { return this; } - public Builder putPlugin(String pluginId, PluginEntry entry) { - this.plugins.put(pluginId, entry); - return this; - } - public Builder buffer(BufferEntry buffer) { this.buffer = buffer != null ? buffer : BufferEntry.empty(); return this;