This is an automated email from the ASF dual-hosted git repository. ramackri pushed a commit to branch ranger-kafka-dynamic-partition-plan in repository https://gitbox.apache.org/repos/asf/ranger.git
commit e9febc95a94d54aad9c3942560e79b96c368795e Author: ramk <[email protected]> AuthorDate: Mon Jun 22 13:18:47 2026 +0530 Remove dead partition-plan helpers and rename PluginScale request model. Drop unused PartitionPlanReplacement and validateReplaceRequest helpers, fix a PMD unused-import warning, and keep compact call formatting for partition-plan service methods. Co-authored-by: Cursor <[email protected]> --- .../kafka/partition/AuthToLocalRuleCatalog.java | 7 ++-- .../kafka/partition/PartitionPlanAllocator.java | 8 ++--- .../partition/PartitionPlanRequestValidator.java | 16 +++------ .../kafka/partition/PartitionPlanService.java | 41 ++++++---------------- .../kafka/partition/ServiceAllowlistBootstrap.java | 13 +++---- .../kafka/partition/model/PartitionPlan.java | 8 ++--- .../partition/model/PartitionPlanReplacement.java | 7 +--- .../partition/model/PluginPartitionAssignment.java | 8 ++--- .../{PluginScaleRequest.java => PluginScale.java} | 4 +-- .../partition/model/ServiceAllowlistEntry.java | 1 - .../org/apache/ranger/audit/rest/AuditREST.java | 7 ++-- .../partition/PartitionPlanReplacementTest.java | 4 +-- .../PartitionPlanServiceMutationTest.java | 10 +++--- 13 files changed, 44 insertions(+), 90 deletions(-) diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/AuthToLocalRuleCatalog.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/AuthToLocalRuleCatalog.java index 054e5cb94..afcba251e 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/AuthToLocalRuleCatalog.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/AuthToLocalRuleCatalog.java @@ -91,8 +91,7 @@ public String composeRulesForAllowedShortNames(Set<String> allowedUserShortNames Set<String> shortNamesMatchedByCatalogRule = new LinkedHashSet<>(); for (PrimaryCatalogRule catalogRule : primaryCatalogRulesInOrder) { - if (catalogRule.mappedShortName != null - && normalizedAllowedShortNames.contains(catalogRule.mappedShortName)) { + if (catalogRule.mappedShortName != null && normalizedAllowedShortNames.contains(catalogRule.mappedShortName)) { composedRuleLines.add(catalogRule.ruleLine); shortNamesMatchedByCatalogRule.add(catalogRule.mappedShortName); } @@ -134,9 +133,7 @@ private static List<String> tokenizeRuleLines(String rawAuthToLocalRules) { } private static boolean isFallbackKerberosRule(String ruleLine) { - return "DEFAULT".equals(ruleLine) - || ruleLine.startsWith("RULE:[1:") - || ruleLine.contains("s/@.*//"); + return "DEFAULT".equals(ruleLine) || ruleLine.startsWith("RULE:[1:") || ruleLine.contains("s/@.*//"); } private static String buildGeneratedShortNameRule(String shortName) { diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanAllocator.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanAllocator.java index d57ca648e..a9456f193 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanAllocator.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanAllocator.java @@ -109,7 +109,7 @@ public static boolean isOnboardAlreadyApplied(PartitionPlan current, String repo return isPromoteAlreadyApplied(current, pluginId, partitionCount, repo, allowedUsers); } - /** Full plan replace (REST PUT) with append-only checks against the current plan. */ + /** 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"); @@ -175,13 +175,9 @@ private static boolean pluginHasPartitionCount(PartitionPlan current, String plu } private static void assertPromoteNotConflicting(PartitionPlan current, String pluginId, int partitionCount, String repo, List<String> allowedUsers) { - if (isPromoteAlreadyApplied(current, pluginId, partitionCount, repo, allowedUsers)) { - return; - } PluginPartitionAssignment existing = Objects.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); + throw new PartitionPlanException("Plugin '" + pluginId + "' already has " + existing.getPartitions().size() + " dedicated partition(s); requested " + partitionCount); } if (StringUtils.isNotBlank(repo)) { ServiceAllowlistEntry serviceEntry = current.getServices().get(repo.trim()); diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanRequestValidator.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanRequestValidator.java index 61bc45074..6229ff1ad 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanRequestValidator.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanRequestValidator.java @@ -25,7 +25,7 @@ import org.apache.ranger.audit.producer.kafka.partition.model.OnboardService; import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlanReplacement; import org.apache.ranger.audit.producer.kafka.partition.model.PromotePlugin; -import org.apache.ranger.audit.producer.kafka.partition.model.PluginScaleRequest; +import org.apache.ranger.audit.producer.kafka.partition.model.PluginScale; import java.util.List; @@ -45,12 +45,6 @@ public static void validatePatchRequest(PartitionPlanReplacement partitionPlanUp } } - /** @deprecated Use {@link #validatePatchRequest(PartitionPlanReplacement)}. */ - @Deprecated - public static void validateReplaceRequest(PartitionPlanReplacement partitionPlanUpdate) { - validatePatchRequest(partitionPlanUpdate); - } - public static void validatePromotePlugin(PromotePlugin promotePluginRequest) { if (promotePluginRequest == null) { throw new PartitionPlanException("Promote plugin request is required"); @@ -65,13 +59,13 @@ public static void validatePromotePlugin(PromotePlugin promotePluginRequest) { } } - public static void validateScalePlugin(String pluginId, PluginScaleRequest scalePluginRequest) { - if (scalePluginRequest == null) { + public static void validateScalePlugin(String pluginId, PluginScale scalePlugin) { + if (scalePlugin == null) { throw new PartitionPlanException("Plugin scale request is required"); } validateNonBlankPluginId(pluginId); - validatePositiveCount(scalePluginRequest.getAdditionalPartitions(), "additionalPartitions"); - validateExpectedVersion(scalePluginRequest.getExpectedVersion()); + validatePositiveCount(scalePlugin.getAdditionalPartitions(), "additionalPartitions"); + validateExpectedVersion(scalePlugin.getExpectedVersion()); } public static void validateOnboardService(OnboardService onboardServiceRequest) { diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanService.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanService.java index 5b6a71ab7..53c31b0f4 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanService.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanService.java @@ -25,7 +25,7 @@ import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlan; import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlanReplacement; import org.apache.ranger.audit.producer.kafka.partition.model.PromotePlugin; -import org.apache.ranger.audit.producer.kafka.partition.model.PluginScaleRequest; +import org.apache.ranger.audit.producer.kafka.partition.model.PluginScale; import org.apache.ranger.audit.provider.MiscUtil; import org.apache.ranger.audit.server.AuditServerConfig; import org.apache.ranger.audit.server.AuditServerConstants; @@ -69,7 +69,7 @@ public PartitionPlan getPartitionPlan() { return plan; } - /** Merges a partial plan delta via REST PUT with optimistic locking. */ + /** Merges a partial plan delta via REST PATCH with optimistic locking. */ public PartitionPlan mergePartitionPlan(PartitionPlanReplacement partitionPlanUpdate, String updatedBy) { PartitionPlanRequestValidator.validatePatchRequest(partitionPlanUpdate); requireDynamicEnabled(); @@ -98,19 +98,10 @@ public PartitionPlan promotePlugin(PromotePlugin promotePluginRequest, String up try (PartitionPlanRegistry registry = registryFactory.open(configProps, INGESTOR_PROP_PREFIX)) { PartitionPlan currentPlan = requirePlan(registry, auditTopic); requireExpectedVersion(currentPlan, promotePluginRequest.getExpectedVersion()); - if (PartitionPlanAllocator.isPromoteAlreadyApplied(currentPlan, - promotePluginRequest.getPluginId(), - promotePluginRequest.getPartitionCount(), - promotePluginRequest.getRepo(), - promotePluginRequest.getAllowedUsers())) { + if (PartitionPlanAllocator.isPromoteAlreadyApplied(currentPlan, promotePluginRequest.getPluginId(), promotePluginRequest.getPartitionCount(), promotePluginRequest.getRepo(), promotePluginRequest.getAllowedUsers())) { return returnCurrentPlanNoOp(currentPlan); } - PartitionPlan nextPlan = PartitionPlanAllocator.promotePlugin(currentPlan, - promotePluginRequest.getPluginId(), - promotePluginRequest.getPartitionCount(), - updatedBy, - promotePluginRequest.getRepo(), - promotePluginRequest.getAllowedUsers()); + PartitionPlan nextPlan = PartitionPlanAllocator.promotePlugin(currentPlan, promotePluginRequest.getPluginId(), promotePluginRequest.getPartitionCount(), updatedBy, promotePluginRequest.getRepo(), promotePluginRequest.getAllowedUsers()); return publishMutation(registry, auditTopic, promotePluginRequest.getExpectedVersion(), currentPlan, nextPlan); } catch (PartitionPlanException e) { throw e; @@ -127,19 +118,10 @@ public PartitionPlan onboardService(OnboardService onboardServiceRequest, String try (PartitionPlanRegistry registry = registryFactory.open(configProps, INGESTOR_PROP_PREFIX)) { PartitionPlan currentPlan = requirePlan(registry, auditTopic); requireExpectedVersion(currentPlan, onboardServiceRequest.getExpectedVersion()); - if (PartitionPlanAllocator.isOnboardAlreadyApplied(currentPlan, - onboardServiceRequest.getServiceName(), - onboardServiceRequest.getPluginId(), - onboardServiceRequest.getPartitionCount(), - onboardServiceRequest.getAllowedUsers())) { + if (PartitionPlanAllocator.isOnboardAlreadyApplied(currentPlan, onboardServiceRequest.getServiceName(), onboardServiceRequest.getPluginId(), onboardServiceRequest.getPartitionCount(), onboardServiceRequest.getAllowedUsers())) { return returnCurrentPlanNoOp(currentPlan); } - PartitionPlan nextPlan = PartitionPlanAllocator.onboardRepo(currentPlan, - onboardServiceRequest.getServiceName(), - onboardServiceRequest.getPluginId(), - onboardServiceRequest.getPartitionCount(), - onboardServiceRequest.getAllowedUsers(), - updatedBy); + PartitionPlan nextPlan = PartitionPlanAllocator.onboardRepo(currentPlan, onboardServiceRequest.getServiceName(), onboardServiceRequest.getPluginId(), onboardServiceRequest.getPartitionCount(), onboardServiceRequest.getAllowedUsers(), updatedBy); return publishMutation(registry, auditTopic, onboardServiceRequest.getExpectedVersion(), currentPlan, nextPlan); } catch (PartitionPlanException e) { throw e; @@ -154,17 +136,14 @@ public Set<String> getPartitionPlanAdminUsers() { } /** Appends tail partitions to a plugin already present in the plan. */ - public PartitionPlan scalePlugin(String pluginId, PluginScaleRequest scalePluginRequest, String updatedBy) { - PartitionPlanRequestValidator.validateScalePlugin(pluginId, scalePluginRequest); + public PartitionPlan scalePlugin(String pluginId, PluginScale scalePlugin, String updatedBy) { + PartitionPlanRequestValidator.validateScalePlugin(pluginId, scalePlugin); requireDynamicEnabled(); String auditTopic = resolveAuditTopicName(); try (PartitionPlanRegistry registry = registryFactory.open(configProps, INGESTOR_PROP_PREFIX)) { PartitionPlan currentPlan = requirePlan(registry, auditTopic); - PartitionPlan nextPlan = PartitionPlanAllocator.scalePlugin(currentPlan, - pluginId, - scalePluginRequest.getAdditionalPartitions(), - updatedBy); - return publishMutation(registry, auditTopic, scalePluginRequest.getExpectedVersion(), currentPlan, nextPlan); + PartitionPlan nextPlan = PartitionPlanAllocator.scalePlugin(currentPlan, pluginId, scalePlugin.getAdditionalPartitions(), updatedBy); + return publishMutation(registry, auditTopic, scalePlugin.getExpectedVersion(), currentPlan, nextPlan); } catch (PartitionPlanException e) { throw e; } catch (Exception e) { diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistBootstrap.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistBootstrap.java index 3478b705b..7f9307c34 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistBootstrap.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/ServiceAllowlistBootstrap.java @@ -47,13 +47,10 @@ public static Map<String, ServiceAllowlistEntry> loadAllowlistsFromProperties(Pr return allowlistEntriesByRepo; } for (String propertyName : ingestorProperties.stringPropertyNames()) { - if (!propertyName.startsWith(PROP_PREFIX_AUDIT_SERVER_SERVICE) - || !propertyName.endsWith(PROP_SUFFIX_ALLOWED_USERS)) { + if (!propertyName.startsWith(PROP_PREFIX_AUDIT_SERVER_SERVICE) || !propertyName.endsWith(PROP_SUFFIX_ALLOWED_USERS)) { continue; } - String serviceRepoName = propertyName.substring( - PROP_PREFIX_AUDIT_SERVER_SERVICE.length(), - propertyName.length() - PROP_SUFFIX_ALLOWED_USERS.length()); + String serviceRepoName = propertyName.substring(PROP_PREFIX_AUDIT_SERVER_SERVICE.length(), propertyName.length() - PROP_SUFFIX_ALLOWED_USERS.length()); if (StringUtils.isBlank(serviceRepoName)) { continue; } @@ -83,12 +80,10 @@ public static Map<String, ServiceAllowlistEntry> loadAllowlistsFromServerConfig( */ public static PartitionPlan mergeSiteXmlAllowlistsWhenPlanServicesMissing( PartitionPlan partitionPlan, Properties ingestorProperties) { - if (partitionPlan == null - || (partitionPlan.getServices() != null && !partitionPlan.getServices().isEmpty())) { + if (partitionPlan == null || (partitionPlan.getServices() != null && !partitionPlan.getServices().isEmpty())) { return partitionPlan; } - Map<String, ServiceAllowlistEntry> siteXmlAllowlistEntries = - loadAllowlistsFromProperties(ingestorProperties); + Map<String, ServiceAllowlistEntry> siteXmlAllowlistEntries = loadAllowlistsFromProperties(ingestorProperties); if (siteXmlAllowlistEntries.isEmpty()) { return partitionPlan; } diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlan.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlan.java index 337a872bf..845555bea 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlan.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlan.java @@ -147,14 +147,14 @@ public static PartitionPlan fromJson(String json) { } @Override - public boolean equals(Object other) { - if (this == other) { + public boolean equals(Object otherPartitionPlanObj) { + if (this == otherPartitionPlanObj) { return true; } - if (other == null || getClass() != other.getClass()) { + if (otherPartitionPlanObj == null || getClass() != otherPartitionPlanObj.getClass()) { return false; } - PartitionPlan otherPartitionPlan = (PartitionPlan) other; + PartitionPlan otherPartitionPlan = (PartitionPlan) otherPartitionPlanObj; return version == otherPartitionPlan.version && topicPartitionCount == otherPartitionPlan.topicPartitionCount && Objects.equals(topic, otherPartitionPlan.topic) diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlanReplacement.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlanReplacement.java index a46f1b8a6..479377c1b 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlanReplacement.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PartitionPlanReplacement.java @@ -36,7 +36,7 @@ import java.util.Map; /** - * Partial partition-plan update (REST PUT). Omitted or empty fields are inherited from the current plan. + * Partial partition-plan update (REST PATCH). Omitted or empty fields are inherited from the current plan. * {@code plugins} and {@code services} merge only entries whose keys are not already present. */ @JsonAutoDetect(getterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE, fieldVisibility = Visibility.ANY) @@ -61,11 +61,6 @@ public PartitionPlanReplacement(@JsonProperty("expectedVersion") int expectedVer this.services = copyServices(services); } - public PartitionPlanReplacement(int expectedVersion, int topicPartitionCount, - Map<String, PluginPartitionAssignment> plugins, PluginPartitionAssignment buffer) { - this(expectedVersion, topicPartitionCount, plugins, buffer, null); - } - public PartitionPlanReplacement(int expectedVersion, int topicPartitionCount, Map<String, PluginPartitionAssignment> plugins, PluginPartitionAssignment buffer, Map<String, ServiceAllowlistEntry> services) { diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PluginPartitionAssignment.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PluginPartitionAssignment.java index 067036f64..50157c9d7 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PluginPartitionAssignment.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PluginPartitionAssignment.java @@ -79,14 +79,14 @@ public int size() { } @Override - public boolean equals(Object other) { - if (this == other) { + public boolean equals(Object otherPluginPartitionAssignmentObj) { + if (this == otherPluginPartitionAssignmentObj) { return true; } - if (other == null || getClass() != other.getClass()) { + if (otherPluginPartitionAssignmentObj == null || getClass() != otherPluginPartitionAssignmentObj.getClass()) { return false; } - PluginPartitionAssignment otherAssignment = (PluginPartitionAssignment) other; + PluginPartitionAssignment otherAssignment = (PluginPartitionAssignment) otherPluginPartitionAssignmentObj; return Objects.equals(partitions, otherAssignment.partitions); } diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PluginScaleRequest.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PluginScale.java similarity index 89% rename from audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PluginScaleRequest.java rename to audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PluginScale.java index d83ceb4dc..73e493f7b 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PluginScaleRequest.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/PluginScale.java @@ -29,12 +29,12 @@ @JsonAutoDetect(getterVisibility = Visibility.NONE, setterVisibility = Visibility.NONE, fieldVisibility = Visibility.ANY) @JsonInclude(JsonInclude.Include.NON_NULL) -public class PluginScaleRequest implements Serializable { +public class PluginScale implements Serializable { private final int additionalPartitions; private final int expectedVersion; @JsonCreator - public PluginScaleRequest(@JsonProperty("additionalPartitions") int additionalPartitions, @JsonProperty("expectedVersion") int expectedVersion) { + public PluginScale(@JsonProperty("additionalPartitions") int additionalPartitions, @JsonProperty("expectedVersion") int expectedVersion) { this.additionalPartitions = additionalPartitions; this.expectedVersion = expectedVersion; } diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/ServiceAllowlistEntry.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/ServiceAllowlistEntry.java index b427b9c4c..b461afde2 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/ServiceAllowlistEntry.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/partition/model/ServiceAllowlistEntry.java @@ -26,7 +26,6 @@ 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; diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/rest/AuditREST.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/rest/AuditREST.java index cdc91f9ea..415e93ed2 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/rest/AuditREST.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/rest/AuditREST.java @@ -32,7 +32,7 @@ import org.apache.ranger.audit.producer.kafka.partition.model.OnboardService; import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlan; import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlanReplacement; -import org.apache.ranger.audit.producer.kafka.partition.model.PluginScaleRequest; +import org.apache.ranger.audit.producer.kafka.partition.model.PluginScale; import org.apache.ranger.audit.producer.kafka.partition.model.PromotePlugin; import org.apache.ranger.audit.provider.MiscUtil; import org.apache.ranger.audit.server.AuditServerConfig; @@ -350,7 +350,7 @@ public Response createPluginAssignment(PromotePlugin request, @Context HttpServl @Path("/partition-plan/plugins/{pluginId}") @Consumes("application/json") @Produces("application/json") - public Response scalePluginAssignment(@PathParam("pluginId") String pluginId, PluginScaleRequest request, @Context HttpServletRequest httpRequest) { + public Response scalePluginAssignment(@PathParam("pluginId") String pluginId, PluginScale pluginScale, @Context HttpServletRequest httpRequest) { LOG.debug("==> AuditREST.scalePluginAssignment(pluginId={})", pluginId); Response ret; if (!partitionPlanService.isDynamicPartitionPlanEnabled()) { @@ -361,8 +361,7 @@ public Response scalePluginAssignment(@PathParam("pluginId") String pluginId, Pl ret = authFailure; } else { try { - ret = toSuccessfulPartitionPlanResponse( - partitionPlanService.scalePlugin(pluginId, request, resolveUpdatedBy(httpRequest))); + ret = toSuccessfulPartitionPlanResponse(partitionPlanService.scalePlugin(pluginId, pluginScale, resolveUpdatedBy(httpRequest))); } catch (PartitionPlanConflictException e) { ret = toPartitionPlanConflictResponse("PATCH /partition-plan/plugins/" + pluginId, e); } catch (PartitionPlanException e) { diff --git a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanReplacementTest.java b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanReplacementTest.java index 09d7155a7..fb294867f 100644 --- a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanReplacementTest.java +++ b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanReplacementTest.java @@ -23,7 +23,7 @@ import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlanReplacement; import org.apache.ranger.audit.producer.kafka.partition.model.PluginPartitionAssignment; import org.apache.ranger.audit.producer.kafka.partition.model.PromotePlugin; -import org.apache.ranger.audit.producer.kafka.partition.model.PluginScaleRequest; +import org.apache.ranger.audit.producer.kafka.partition.model.PluginScale; import org.apache.ranger.audit.producer.kafka.partition.model.ServiceAllowlistEntry; import org.junit.jupiter.api.Test; @@ -149,6 +149,6 @@ public void testRequestValidatorRejectsOnboardServiceWithoutAllowedUsers() { @Test public void testRequestValidatorRejectsScaleWithZeroAdditionalPartitions() { assertThrows(PartitionPlanException.class, - () -> PartitionPlanRequestValidator.validateScalePlugin("hiveServer2", new PluginScaleRequest(0, 1))); + () -> PartitionPlanRequestValidator.validateScalePlugin("hiveServer2", new PluginScale(0, 1))); } } diff --git a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanServiceMutationTest.java b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanServiceMutationTest.java index b272e58a9..ce2ad8b56 100644 --- a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanServiceMutationTest.java +++ b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/partition/PartitionPlanServiceMutationTest.java @@ -24,7 +24,7 @@ import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlanReplacement; import org.apache.ranger.audit.producer.kafka.partition.model.PluginPartitionAssignment; import org.apache.ranger.audit.producer.kafka.partition.model.PromotePlugin; -import org.apache.ranger.audit.producer.kafka.partition.model.PluginScaleRequest; +import org.apache.ranger.audit.producer.kafka.partition.model.PluginScale; import org.apache.ranger.audit.producer.kafka.partition.model.ServiceAllowlistEntry; import org.apache.ranger.audit.server.AuditServerConstants; import org.junit.jupiter.api.AfterEach; @@ -96,7 +96,7 @@ public void testScalePluginAppendsTailPartitionsViaDedicatedEndpoint() throws Ex MutableRegistry registry = new MutableRegistry(initialPlan); PartitionPlanService service = service(registry, new NoOpAuditTopicPartitionGrower()); - PartitionPlan result = service.scalePlugin("hiveServer2", new PluginScaleRequest(2, 1), "ops"); + PartitionPlan result = service.scalePlugin("hiveServer2", new PluginScale(2, 1), "ops"); assertEquals(2, result.getVersion()); assertEquals(17, result.getTopicPartitionCount()); @@ -151,7 +151,7 @@ public void testScalePluginRejectsInvalidAdditionalPartitions() { PartitionPlanService service = service(registry, new NoOpAuditTopicPartitionGrower()); PartitionPlanException error = assertThrows(PartitionPlanException.class, - () -> service.scalePlugin("hiveServer2", new PluginScaleRequest(0, 1), "ops")); + () -> service.scalePlugin("hiveServer2", new PluginScale(0, 1), "ops")); assertTrue(error.getMessage().contains("additionalPartitions")); } @@ -318,10 +318,10 @@ public void testScalePluginStillAppendsOnRepeat() throws Exception { MutableRegistry registry = new MutableRegistry(initialPlan); PartitionPlanService service = service(registry, new NoOpAuditTopicPartitionGrower()); - service.scalePlugin("hiveServer2", new PluginScaleRequest(2, 1), "ops"); + service.scalePlugin("hiveServer2", new PluginScale(2, 1), "ops"); assertEquals(1, registry.getWriteCount()); - PartitionPlan second = service.scalePlugin("hiveServer2", new PluginScaleRequest(2, 2), "ops"); + PartitionPlan second = service.scalePlugin("hiveServer2", new PluginScale(2, 2), "ops"); assertEquals(3, second.getVersion()); assertEquals(2, registry.getWriteCount());
