You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2026/09/23 08:02:18 UTC
(kafka) branch trunk updated: KAFKA-20774 Fix ScramDelta.apply() dropping SCRAM users (#22763)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 14cde029c53 KAFKA-20774 Fix ScramDelta.apply() dropping SCRAM users (#22763)
14cde029c53 is described below
commit 14cde029c53cb7d7094cc4dd86e1c1607c0ab0af
Author: dark231321 <49...@users.noreply.github.com>
AuthorDate: Wed Sep 23 12:02:11 2026 +0400
KAFKA-20774 Fix ScramDelta.apply() dropping SCRAM users (#22763)
Fixes: https://issues.apache.org/jira/browse/KAFKA-20774
Defer removing an empty per-mechanism user map until all changes for
that mechanism have been applied in `ScramDelta.apply()`.
Reviewers: Chia-Ping Tsai <ch...@gmail.com>, Kevin Wu
<kw...@confluent.io>, José Armando García Sancio <js...@apache.org>
---
.../java/org/apache/kafka/image/ScramDelta.java | 6 +-
.../org/apache/kafka/image/ScramDeltaTest.java | 118 +++++++++++++++++++++
2 files changed, 121 insertions(+), 3 deletions(-)
diff --git a/metadata/src/main/java/org/apache/kafka/image/ScramDelta.java b/metadata/src/main/java/org/apache/kafka/image/ScramDelta.java
index bce7ece0c49..706dc728ef4 100644
--- a/metadata/src/main/java/org/apache/kafka/image/ScramDelta.java
+++ b/metadata/src/main/java/org/apache/kafka/image/ScramDelta.java
@@ -92,11 +92,11 @@ public final class ScramDelta {
userMap.put(userNameEntry.getKey(), userNameEntry.getValue().get());
} else {
userMap.remove(userNameEntry.getKey());
- if (userMap.isEmpty()) {
- newMechanisms.remove(mechanismChangeEntry.getKey());
- }
}
}
+ if (userMap.isEmpty()) {
+ newMechanisms.remove(mechanismChangeEntry.getKey());
+ }
}
return new ScramImage(newMechanisms);
}
diff --git a/metadata/src/test/java/org/apache/kafka/image/ScramDeltaTest.java b/metadata/src/test/java/org/apache/kafka/image/ScramDeltaTest.java
new file mode 100644
index 00000000000..ec9be713f32
--- /dev/null
+++ b/metadata/src/test/java/org/apache/kafka/image/ScramDeltaTest.java
@@ -0,0 +1,118 @@
+/*
+ * 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.kafka.image;
+
+import org.apache.kafka.clients.admin.ScramMechanism;
+import org.apache.kafka.common.metadata.RemoveUserScramCredentialRecord;
+import org.apache.kafka.metadata.ScramCredentialData;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.kafka.clients.admin.ScramMechanism.SCRAM_SHA_256;
+import static org.apache.kafka.clients.admin.ScramMechanism.SCRAM_SHA_512;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Regression tests for {@link ScramDelta#apply()}.
+ *
+ * <p>Absent REMOVE records must not detach the per-mechanism user map before later UPSERTs
+ * in the same batch are applied.
+ */
+@Timeout(value = 40)
+public class ScramDeltaTest {
+
+ @Test
+ public void testApplyWithNoChangesReturnsBaseImage() {
+ ScramImage base = image(SCRAM_SHA_256, "user-admin");
+ assertEquals(base, new ScramDelta(base).apply());
+ }
+
+ @Test
+ public void testAbsentRemovesBeforeUpsertPreservesUsers() {
+ ScramDelta delta = new ScramDelta(ScramImage.EMPTY);
+ replayRemove(delta, "producer");
+ replayRemove(delta, "consumer");
+ replayUpsert(delta, SCRAM_SHA_256, "user-admin");
+ replayRemove(delta, "update-user");
+ replayRemove(delta, "ext-user");
+ replayUpsert(delta, SCRAM_SHA_512, "cluster-admin");
+
+ ScramImage result = delta.apply();
+
+ assertUserNames(result, SCRAM_SHA_256, "user-admin");
+ assertUserNames(result, SCRAM_SHA_512, "cluster-admin");
+ }
+
+ @Test
+ public void testAbsentRemovesOnlyLeaveImageEmpty() {
+ ScramDelta delta = new ScramDelta(ScramImage.EMPTY);
+ replayRemove(delta, "producer");
+ replayRemove(delta, "consumer");
+
+ assertTrue(delta.apply().isEmpty());
+ }
+
+ @Test
+ public void testRemoveLastExistingUserClearsImage() {
+ ScramImage base = image(SCRAM_SHA_256, "user-admin");
+ ScramDelta delta = new ScramDelta(base);
+ replayRemove(delta, "user-admin");
+
+ assertTrue(delta.apply().isEmpty());
+ }
+
+ private static ScramImage image(ScramMechanism mechanism, String... userNames) {
+ Map<String, ScramCredentialData> users = Arrays.stream(userNames)
+ .collect(Collectors.toMap(name -> name, ScramDeltaTest::credential));
+ return new ScramImage(Map.of(mechanism, users));
+ }
+
+ private static ScramCredentialData credential(String userName) {
+ int seed = userName.hashCode();
+ byte[] bytes = new byte[] {
+ (byte) seed,
+ (byte) (seed >> 8),
+ (byte) (seed >> 16)
+ };
+ return new ScramCredentialData(bytes, bytes, bytes, 4096);
+ }
+
+ private static void replayUpsert(ScramDelta delta, ScramMechanism mechanism, String userName) {
+ ScramCredentialData data = credential(userName);
+ delta.replay(data.toRecord(userName, mechanism));
+ }
+
+ private static void replayRemove(ScramDelta delta, String userName) {
+ delta.replay(new RemoveUserScramCredentialRecord()
+ .setName(userName)
+ .setMechanism(ScramMechanism.SCRAM_SHA_256.type()));
+ }
+
+ private static void assertUserNames(ScramImage image, ScramMechanism mechanism, String... userNames) {
+ assertEquals(
+ Arrays.stream(userNames).collect(Collectors.toSet()),
+ image.mechanisms().get(mechanism).keySet()
+ );
+ }
+}