You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "emissionnebula (via GitHub)" <gi...@apache.org> on 2023/03/22 20:26:45 UTC
[GitHub] [kafka] emissionnebula opened a new pull request, #13437: KAFKA-14828: Remove R/W locks using persistent data structures
emissionnebula opened a new pull request, #13437:
URL: https://github.com/apache/kafka/pull/13437
Currently, StandardAuthorizer uses a R/W lock for maintaining the consistency of data. For the clusters with very high traffic, we will typically see an increase in latencies whenever a write operation comes. The intent of this PR is to get rid of the R/W lock with the help of immutable or persistent collections. Basically, new object references are used to hold the intermediate state of the write operation. After the completion of the operation, the main reference to the cache is changed to point to the new object. Also, for the read operation, the code is changed such that all accesses to the cache for a single read operation are done to a particular cache object only.
In another [PR](https://github.com/apache/kafka/pull/13280), new persistent libraries ([Paguro](https://github.com/GlenKPeterson/Paguro/) and [Vavr](https://github.com/vavr-io/vavr)) have been evaluated to improve the performance of Kraft metadata image change performance. That PR favors Vavr over Paguro.
In the description below, you can find the performance of various libraries at the time of both read and write. Read performance is checked with the existing `AuthorizerBenchmark`. For write performance, a new `AuthorizerUpdateBenchmark` has been added which evaluates the performance of the `addAcl` operation.
# Library Comparison (TLDR)
| | PCollections | Paguro | Vavr |
|--------|--------|--------|--------|
| Implements Java native collection interfaces | ✅ | ✅ | ❗ (Provides adaptors to convert to Java, which in case of TreeSet are not efficient) |
| Read Performance | ❌ (10000x of AclAuthorizer) | ✅ (Similar to AclAuthorizer) | ❌ (1000x of AclAuthorizer) |
| Write Performance | ⚠️ (10x of Paguro) | ✅ (best) | ✅ (4x of Paguro) |
| Library Completeness | ✅ (HashMap, HashSet, TreeMap, TreeSet, OrderList, OrderedSet, OrderedMap, Stack, Queue) | ⚠️ (HashMap, HashSet, TreeMap, TreeSet, List) | ✅ (most complete - Array, List, Vector, HashMap, HashSet, LinkedHashMap, LinkedHashSet, PriorityQueue, TreeMap, TreeSet) |
# Conclusion
Since the read time performance is better for pcollections compared to Vavr or Paguro, for StandardAuthorizer it makes sense to go ahead with it. But for Kraft metadata image change patch, `pcollections` was [not considered](https://github.com/apache/kafka/pull/13280#issuecomment-1442082585) because of performance concerns explained on this [page](https://github.com/GlenKPeterson/Paguro/wiki/UncleJim-vs.-PCollections).
So now there are two ways forward -
1. Check the performance of Kafka metadata image change with pcollections and see if we are also noticing any difference compared to Vavr.
2. Include two separate libraries in Kafka - `pcollections` for StandardAuthorizer and Vavr for Kraft metadata image change.
# Appendix: Detailed Library Comparison
## Current Performance
### AclAuthorizer Read
```
Benchmark (aclCount) (denyPercentage) (resourceCount) Mode Cnt Score Error Units
AuthorizerBenchmark.testAclsIterator 50 20 50000 avgt 2 186.399 ms/op
AuthorizerBenchmark.testAuthorizeByResourceType 50 20 50000 avgt 2 0.005 ms/op
AuthorizerBenchmark.testAuthorizer 50 20 50000 avgt 2 0.244 ms/op
AuthorizerBenchmark.testUpdateCache 50 20 50000 avgt 2 502.015 ms/op
JMH benchmarks done
```
### AclAuthorizer Read
```
Benchmark (aclCount) (denyPercentage) (resourceCount) Mode Cnt Score Error Units
AuthorizerBenchmark.testAclsIterator 50 20 50000 avgt 2 218.633 ms/op
AuthorizerBenchmark.testAuthorizeByResourceType 50 20 50000 avgt 2 486.814 ms/op
AuthorizerBenchmark.testAuthorizer 50 20 50000 avgt 2 0.750 ms/op
AuthorizerBenchmark.testUpdateCache 50 20 50000 avgt 2 ≈ 10⁻⁶ ms/op
JMH benchmarks done
```
## PCollections
### Read
```
Benchmark (aclCount) (denyPercentage) (resourceCount) Mode Cnt Score Error Units
AuthorizerBenchmark.testAclsIterator 50 20 50000 avgt 2 208.878 ms/op
AuthorizerBenchmark.testAuthorizeByResourceType 50 20 50000 avgt 2 509.253 ms/op
AuthorizerBenchmark.testAuthorizer 50 20 50000 avgt 2 0.785 ms/op
AuthorizerBenchmark.testUpdateCache 50 20 50000 avgt 2 ≈ 10⁻⁶ ms/op
JMH benchmarks done
```
### Write
```
Benchmark Mode Cnt Score Error Units
AuthorizerUpdateBenchmark.testAddAcl avgt 4 9.751 ± 2.633 ms/op
JMH benchmarks done
```
## Paguro
⚠️ Note - There are two bugs in the TreeSet code of Paguro. Following results are computed after fixing those bugs in local.
### Read
```
Benchmark (aclCount) (denyPercentage) (resourceCount) Mode Cnt Score Error Units
AuthorizerBenchmark.testAclsIterator 50 20 50000 avgt 2 289.411 ms/op
AuthorizerBenchmark.testAuthorizeByResourceType 50 20 50000 avgt 2 1048.940 ms/op
AuthorizerBenchmark.testAuthorizer 50 20 50000 avgt 2 3751.478 ms/op
AuthorizerBenchmark.testUpdateCache 50 20 50000 avgt 2 ≈ 10⁻⁶ ms/op
JMH benchmarks done
```
⚠️ - Read performance of Paguro is worst than pcollections because of inefficient implementation of the `tailSet()` function. In Paguro, it is an O(n) operation however in `pcollections` it is an O(log(n)) operation.
### Write
```Benchmark Mode Cnt Score Error Units
AuthorizerUpdateBenchmark.testAddAcl avgt 4 0.509 ± 1.108 ms/op
JMH benchmarks done
```
## Vavr
### Read
```
Vavr
Benchmark (aclCount) (denyPercentage) (resourceCount) Mode Cnt Score Error Units
AuthorizerBenchmark.testAclsIterator 50 20 50000 avgt 2 277.919 ms/op
AuthorizerBenchmark.testAuthorizeByResourceType 50 20 50000 avgt 2 3406.584 ms/op
AuthorizerBenchmark.testAuthorizer 50 20 50000 avgt 2 3465.406 ms/op
AuthorizerBenchmark.testUpdateCache 50 20 50000 avgt 2 ≈ 10⁻⁶ ms/op
JMH benchmarks done
```
⚠️ - Vavr doesn't implement Java's SortedSet interface and it doesn't expose any API equivalent to [SortedSet.tailSet()](https://docs.oracle.com/javase/8/docs/api/java/util/SortedSet.html#tailSet-E-). So, it is required to convert the vavr's TreeSet into Java equivalent, which is an O(n) operation, and then call `tailSet` over it.
### Write
```
Benchmark Mode Cnt Score Error Units
AuthorizerUpdateBenchmark.testAddAcl avgt 4 1.515 ± 0.995 ms/op
JMH benchmarks done
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a diff in pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on code in PR #13437:
URL: https://github.com/apache/kafka/pull/13437#discussion_r1172661731
##########
server-common/src/main/java/org/apache/kafka/server/immutable/ImmutableNavigableSet.java:
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.server.immutable;
+
+import org.apache.kafka.server.immutable.pcollections.PCollectionsImmutableNavigableSet;
+
+import java.util.NavigableSet;
+
+/**
+ * A persistent Tree-based NavigableSet wrapper
+ * java.util.Set methods that mutate in-place will throw UnsupportedOperationException
+ *
+ * @param <E> the element type
+ */
+public interface ImmutableNavigableSet<E> extends ImmutableSet<E>, NavigableSet<E> {
+ /**
+ * @return a wrapped tree-based persistent navigable set that is empty
+ * @param <E> the element type
+ */
+ static <E extends Comparable<? super E>> ImmutableNavigableSet<E> empty() {
+ return PCollectionsImmutableNavigableSet.empty();
+ }
+
+ /**
+ * @param e the element
+ * @return a wrapped hash-based persistent set that is empty
+ * @param <E> the element type
+ */
+ static <E extends Comparable<? super E>> ImmutableNavigableSet<E> singleton(E e) {
+ return PCollectionsImmutableNavigableSet.singleton(e);
+ }
+
+ /**
+ * @param e the element
+ * @return a wrapped persistent sorted set that differs from this one in that the given element is added (if necessary)
+ */
+ ImmutableNavigableSet<E> added(E e);
+
+ /**
+ * @param e the element
+ * @return a wrapped persistent sorted set that differs from this one in that the given element is added (if necessary)
Review Comment:
s/sorted/navigable/
##########
server-common/src/main/java/org/apache/kafka/server/immutable/ImmutableNavigableSet.java:
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.server.immutable;
+
+import org.apache.kafka.server.immutable.pcollections.PCollectionsImmutableNavigableSet;
+
+import java.util.NavigableSet;
+
+/**
+ * A persistent Tree-based NavigableSet wrapper
+ * java.util.Set methods that mutate in-place will throw UnsupportedOperationException
+ *
+ * @param <E> the element type
+ */
+public interface ImmutableNavigableSet<E> extends ImmutableSet<E>, NavigableSet<E> {
+ /**
+ * @return a wrapped tree-based persistent navigable set that is empty
+ * @param <E> the element type
+ */
+ static <E extends Comparable<? super E>> ImmutableNavigableSet<E> empty() {
+ return PCollectionsImmutableNavigableSet.empty();
+ }
+
+ /**
+ * @param e the element
+ * @return a wrapped hash-based persistent set that is empty
Review Comment:
Looks like copy/paste error -- not hash-based.
##########
server-common/src/test/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableNavigableSetTest.java:
##########
@@ -0,0 +1,422 @@
+/*
+ * 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.server.immutable.pcollections;
+
+import org.apache.kafka.server.immutable.DelegationChecker;
+import org.apache.kafka.server.immutable.ImmutableNavigableSet;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.pcollections.HashTreePSet;
+import org.pcollections.TreePSet;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.Random;
+import java.util.Spliterator;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.stream.Stream;
+
+import static java.util.function.Function.identity;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+
+@SuppressWarnings({"unchecked", "deprecation"})
+public class PCollectionsImmutableNavigableSetTest {
+
+ private static final TreePSet<Integer> SINGLETON_SET = TreePSet.singleton(new Random().nextInt());
+
+ private static final class PCollectionsTreeSetWrapperDelegationChecker<R> extends DelegationChecker<TreePSet<Object>, PCollectionsImmutableNavigableSet<Object>, R> {
+ public PCollectionsTreeSetWrapperDelegationChecker() {
+ super(mock(TreePSet.class), PCollectionsImmutableNavigableSet::new);
+ }
+
+ public TreePSet<Object> unwrap(PCollectionsImmutableNavigableSet<Object> wrapper) {
+ return wrapper.underlying();
+ }
+ }
+
+ @Test
+ public void testEmptySet() {
+ Assertions.assertEquals(HashTreePSet.empty(), ((PCollectionsImmutableNavigableSet<?>) ImmutableNavigableSet.empty()).underlying());
+ }
+
+ @Test
+ public void testSingletonSet() {
+ Assertions.assertEquals(HashTreePSet.singleton(1), ((PCollectionsImmutableNavigableSet<?>) ImmutableNavigableSet.singleton(1)).underlying());
+ }
+
+ @Test
+ public void testUnderlying() {
+ assertSame(SINGLETON_SET, new PCollectionsImmutableNavigableSet<>(SINGLETON_SET).underlying());
+ }
+
+ @Test
+ public void testDelegationOfAfterAdding() {
Review Comment:
My oversight; I forgot to rename these methods in the test classes when we renamed the methods on the interfaces.
Can you rename this `testDelegationOfAdded()` and rename `testDelegationOfAfterRemoving()` to `testDelegationOfRemoved()` -- both here and in `PCollectionsImmutableSetTest`? And then do a similar rename in `PCollectionsImmutableMapTest`?
##########
server-common/src/test/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableMapTest.java:
##########
@@ -225,77 +225,59 @@ public void testDelegationOfReplaceAll() {
new PCollectionsHashMapWrapperDelegationChecker<>()
.defineMockConfigurationForVoidMethodInvocation(mock -> mock.replaceAll(eq(mockBiFunction)))
.defineWrapperVoidMethodInvocation(wrapper -> wrapper.replaceAll(mockBiFunction))
- .doVoidMethodDelegationCheck();
+ .doUnsupportedVoidFunctionDelegrationCheck();
}
@Test
public void testDelegationOfPutIfAbsent() {
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.putIfAbsent(eq(this), eq(this)), this)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.putIfAbsent(this, this), identity())
- .doFunctionDelegationCheck();
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testDelegationOfRemoveByKeyAndValue(boolean mockFunctionReturnValue) {
- new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.remove(eq(this), eq(this)), mockFunctionReturnValue)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.remove(this, this), identity())
- .doFunctionDelegationCheck();
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testDelegationOfReplaceWhenMappedToSpecificValue(boolean mockFunctionReturnValue) {
- new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.replace(eq(this), eq(this), eq(this)), mockFunctionReturnValue)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.replace(this, this, this), identity())
- .doFunctionDelegationCheck();
+ .defineMockConfigurationForUnsupportedFunction(mock -> mock.putIfAbsent(eq(this), eq(this)))
+ .defineWrapperUnsupportedFunctionInvocation(wrapper -> wrapper.putIfAbsent(this, this))
+ .doUnsupportedFunctionDelegrationCheck();
}
@Test
public void testDelegationOfReplaceWhenMappedToAnyValue() {
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.replace(eq(this), eq(this)), this)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.replace(this, this), identity())
- .doFunctionDelegationCheck();
+ .defineMockConfigurationForUnsupportedFunction(mock -> mock.replace(eq(this), eq(this)))
+ .defineWrapperUnsupportedFunctionInvocation(wrapper -> wrapper.replace(this, this))
+ .doUnsupportedFunctionDelegrationCheck();
}
@Test
public void testDelegationOfComputeIfAbsent() {
final Function<Object, Object> mockFunction = mock(Function.class);
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForVoidMethodInvocation(mock -> mock.computeIfAbsent(eq(this), eq(mockFunction)))
- .defineWrapperVoidMethodInvocation(wrapper -> wrapper.computeIfAbsent(this, mockFunction))
- .doVoidMethodDelegationCheck();
+ .defineMockConfigurationForUnsupportedFunction(mock -> mock.computeIfAbsent(eq(this), eq(mockFunction)))
+ .defineWrapperUnsupportedFunctionInvocation(wrapper -> wrapper.computeIfAbsent(this, mockFunction))
+ .doUnsupportedFunctionDelegrationCheck();
}
@Test
public void testDelegationOfComputeIfPresent() {
final BiFunction<Object, Object, Object> mockBiFunction = mock(BiFunction.class);
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForVoidMethodInvocation(mock -> mock.computeIfPresent(eq(this), eq(mockBiFunction)))
- .defineWrapperVoidMethodInvocation(wrapper -> wrapper.computeIfPresent(this, mockBiFunction))
- .doVoidMethodDelegationCheck();
+ .defineMockConfigurationForUnsupportedFunction(mock -> mock.computeIfPresent(eq(this), eq(mockBiFunction)))
+ .defineWrapperUnsupportedFunctionInvocation(wrapper -> wrapper.computeIfPresent(this, mockBiFunction))
+ .doUnsupportedFunctionDelegrationCheck();
Review Comment:
Nice catch on fixing this; computeIfPresent() is a function rather than a void method.
##########
server-common/src/test/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableMapTest.java:
##########
@@ -225,77 +225,59 @@ public void testDelegationOfReplaceAll() {
new PCollectionsHashMapWrapperDelegationChecker<>()
.defineMockConfigurationForVoidMethodInvocation(mock -> mock.replaceAll(eq(mockBiFunction)))
.defineWrapperVoidMethodInvocation(wrapper -> wrapper.replaceAll(mockBiFunction))
- .doVoidMethodDelegationCheck();
+ .doUnsupportedVoidFunctionDelegrationCheck();
}
@Test
public void testDelegationOfPutIfAbsent() {
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.putIfAbsent(eq(this), eq(this)), this)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.putIfAbsent(this, this), identity())
- .doFunctionDelegationCheck();
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testDelegationOfRemoveByKeyAndValue(boolean mockFunctionReturnValue) {
- new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.remove(eq(this), eq(this)), mockFunctionReturnValue)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.remove(this, this), identity())
- .doFunctionDelegationCheck();
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testDelegationOfReplaceWhenMappedToSpecificValue(boolean mockFunctionReturnValue) {
- new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.replace(eq(this), eq(this), eq(this)), mockFunctionReturnValue)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.replace(this, this, this), identity())
- .doFunctionDelegationCheck();
+ .defineMockConfigurationForUnsupportedFunction(mock -> mock.putIfAbsent(eq(this), eq(this)))
+ .defineWrapperUnsupportedFunctionInvocation(wrapper -> wrapper.putIfAbsent(this, this))
+ .doUnsupportedFunctionDelegrationCheck();
}
@Test
public void testDelegationOfReplaceWhenMappedToAnyValue() {
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.replace(eq(this), eq(this)), this)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.replace(this, this), identity())
- .doFunctionDelegationCheck();
+ .defineMockConfigurationForUnsupportedFunction(mock -> mock.replace(eq(this), eq(this)))
+ .defineWrapperUnsupportedFunctionInvocation(wrapper -> wrapper.replace(this, this))
+ .doUnsupportedFunctionDelegrationCheck();
}
@Test
public void testDelegationOfComputeIfAbsent() {
final Function<Object, Object> mockFunction = mock(Function.class);
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForVoidMethodInvocation(mock -> mock.computeIfAbsent(eq(this), eq(mockFunction)))
- .defineWrapperVoidMethodInvocation(wrapper -> wrapper.computeIfAbsent(this, mockFunction))
- .doVoidMethodDelegationCheck();
+ .defineMockConfigurationForUnsupportedFunction(mock -> mock.computeIfAbsent(eq(this), eq(mockFunction)))
+ .defineWrapperUnsupportedFunctionInvocation(wrapper -> wrapper.computeIfAbsent(this, mockFunction))
+ .doUnsupportedFunctionDelegrationCheck();
}
@Test
public void testDelegationOfComputeIfPresent() {
final BiFunction<Object, Object, Object> mockBiFunction = mock(BiFunction.class);
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForVoidMethodInvocation(mock -> mock.computeIfPresent(eq(this), eq(mockBiFunction)))
- .defineWrapperVoidMethodInvocation(wrapper -> wrapper.computeIfPresent(this, mockBiFunction))
- .doVoidMethodDelegationCheck();
+ .defineMockConfigurationForUnsupportedFunction(mock -> mock.computeIfPresent(eq(this), eq(mockBiFunction)))
+ .defineWrapperUnsupportedFunctionInvocation(wrapper -> wrapper.computeIfPresent(this, mockBiFunction))
+ .doUnsupportedFunctionDelegrationCheck();
}
@Test
public void testDelegationOfCompute() {
final BiFunction<Object, Object, Object> mockBiFunction = mock(BiFunction.class);
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForVoidMethodInvocation(mock -> mock.compute(eq(this), eq(mockBiFunction)))
- .defineWrapperVoidMethodInvocation(wrapper -> wrapper.compute(this, mockBiFunction))
- .doVoidMethodDelegationCheck();
+ .defineMockConfigurationForUnsupportedFunction(mock -> mock.compute(eq(this), eq(mockBiFunction)))
+ .defineWrapperUnsupportedFunctionInvocation(wrapper -> wrapper.compute(this, mockBiFunction))
+ .doUnsupportedFunctionDelegrationCheck();
Review Comment:
Nice catch on fixing this; compute() is a function rather than a void method.
##########
server-common/src/test/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableMapTest.java:
##########
@@ -225,77 +225,59 @@ public void testDelegationOfReplaceAll() {
new PCollectionsHashMapWrapperDelegationChecker<>()
.defineMockConfigurationForVoidMethodInvocation(mock -> mock.replaceAll(eq(mockBiFunction)))
.defineWrapperVoidMethodInvocation(wrapper -> wrapper.replaceAll(mockBiFunction))
- .doVoidMethodDelegationCheck();
+ .doUnsupportedVoidFunctionDelegrationCheck();
}
@Test
public void testDelegationOfPutIfAbsent() {
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.putIfAbsent(eq(this), eq(this)), this)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.putIfAbsent(this, this), identity())
- .doFunctionDelegationCheck();
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testDelegationOfRemoveByKeyAndValue(boolean mockFunctionReturnValue) {
- new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.remove(eq(this), eq(this)), mockFunctionReturnValue)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.remove(this, this), identity())
- .doFunctionDelegationCheck();
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testDelegationOfReplaceWhenMappedToSpecificValue(boolean mockFunctionReturnValue) {
- new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.replace(eq(this), eq(this), eq(this)), mockFunctionReturnValue)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.replace(this, this, this), identity())
- .doFunctionDelegationCheck();
+ .defineMockConfigurationForUnsupportedFunction(mock -> mock.putIfAbsent(eq(this), eq(this)))
+ .defineWrapperUnsupportedFunctionInvocation(wrapper -> wrapper.putIfAbsent(this, this))
+ .doUnsupportedFunctionDelegrationCheck();
}
@Test
public void testDelegationOfReplaceWhenMappedToAnyValue() {
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.replace(eq(this), eq(this)), this)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.replace(this, this), identity())
- .doFunctionDelegationCheck();
+ .defineMockConfigurationForUnsupportedFunction(mock -> mock.replace(eq(this), eq(this)))
+ .defineWrapperUnsupportedFunctionInvocation(wrapper -> wrapper.replace(this, this))
+ .doUnsupportedFunctionDelegrationCheck();
}
@Test
public void testDelegationOfComputeIfAbsent() {
final Function<Object, Object> mockFunction = mock(Function.class);
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForVoidMethodInvocation(mock -> mock.computeIfAbsent(eq(this), eq(mockFunction)))
- .defineWrapperVoidMethodInvocation(wrapper -> wrapper.computeIfAbsent(this, mockFunction))
- .doVoidMethodDelegationCheck();
+ .defineMockConfigurationForUnsupportedFunction(mock -> mock.computeIfAbsent(eq(this), eq(mockFunction)))
+ .defineWrapperUnsupportedFunctionInvocation(wrapper -> wrapper.computeIfAbsent(this, mockFunction))
+ .doUnsupportedFunctionDelegrationCheck();
Review Comment:
Nice catch on fixing this; computeIfAbsent() is a function rather than a void method.
##########
server-common/src/test/java/org/apache/kafka/server/immutable/DelegationChecker.java:
##########
@@ -139,6 +163,16 @@ public void doFunctionDelegationCheck() {
}
}
+ public void doUnsupportedFunctionDelegrationCheck() {
Review Comment:
`s/Delegration/Delegation/` (and same above in `doUnsupportedVoidFunctionDelegrationCheck()`).
I think what is going on with these methods you added is that you are configuring the mock to invoke the real `TreePSet` method and then confirming that the invocation of the real method throws `UnsupportedOperationException`. The methods that (for example) `PCollectionsImmutableMapTest.testDelegationOfPut()` were previously invoking were not invoking the underlying `HashPMap` method but were instead configuring the mock to flag the fact that the delegation occurred as expected (by setting `persistentCollectionMethodInvokedCorrectly = true`).
I think on the one hand we don't have to test that the real method throws the exception -- that's up to the library itself to test in its test suite. All we have to test is that we delegate correctly. This is why I implemented it this way. However, I think it is useful to explicitly call out the fact that these methods will throw that exception. So I'm fine with this change in general. We just have to correct the `Delegration` misspelling. Thanks for the improvement.
What do you think about adjusting the test method names where we expect the method to throw the exception? So a test method might be `testDelegationOfUnsupportedOperationAdd()`? I think this might be a good change.
##########
server-common/src/main/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableNavigableSet.java:
##########
@@ -0,0 +1,273 @@
+/*
+ * 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.server.immutable.pcollections;
+
+import org.apache.kafka.server.immutable.ImmutableNavigableSet;
+import org.pcollections.TreePSet;
+
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.Objects;
+import java.util.Spliterator;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.stream.Stream;
+
+@SuppressWarnings("deprecation")
+public class PCollectionsImmutableNavigableSet<E> implements ImmutableNavigableSet<E> {
+ private final TreePSet<E> underlying;
+
+ /**
+ * @return a wrapped tree-based persistent navigable set that is empty
+ * @param <E> the element type
+ */
+ public static <E extends Comparable<? super E>> PCollectionsImmutableNavigableSet<E> empty() {
+ return new PCollectionsImmutableNavigableSet<>(TreePSet.<E>empty());
+ }
+
+ /**
+ * @param e the element
+ * @return a wrapped hash-based persistent set that is empty
+ * @param <E> the element type
+ */
+ public static <E extends Comparable<? super E>> PCollectionsImmutableNavigableSet<E> singleton(E e) {
+ return new PCollectionsImmutableNavigableSet<>(TreePSet.singleton(e));
+ }
+
+ public PCollectionsImmutableNavigableSet(TreePSet<E> underlying) {
+ this.underlying = underlying;
+ }
+
+ @Override
+ public PCollectionsImmutableNavigableSet<E> added(E e) {
+ return new PCollectionsImmutableNavigableSet<>(underlying.plus(e));
+ }
+
+ @Override
+ public PCollectionsImmutableNavigableSet<E> removed(E e) {
+ return new PCollectionsImmutableNavigableSet<>(underlying.minus(e));
+ }
+
+ @Override
+ public E lower(E e) {
+ return underlying.lower(e);
+ }
+
+ @Override
+ public E floor(E e) {
+ return underlying.floor(e);
+ }
+
+ @Override
+ public E ceiling(E e) {
+ return underlying.ceiling(e);
+ }
+
+ @Override
+ public E higher(E e) {
+ return underlying.higher(e);
+ }
+
+ @Override
+ public E pollFirst() {
+ // will throw UnsupportedOperationException
+ return underlying.pollFirst();
+ }
+
+ @Override
+ public E pollLast() {
+ // will throw UnsupportedOperationException
+ return underlying.pollLast();
+ }
+
+ @Override
+ public PCollectionsImmutableNavigableSet<E> descendingSet() {
+ return new PCollectionsImmutableNavigableSet<>(underlying.descendingSet());
+ }
+
+ @Override
+ public Iterator<E> descendingIterator() {
+ return underlying.descendingIterator();
+ }
+
+ @Override
+ public PCollectionsImmutableNavigableSet<E> subSet(E fromElement, boolean fromInclusive, E toElement, boolean toInclusive) {
+ return new PCollectionsImmutableNavigableSet<>(underlying.subSet(fromElement, fromInclusive, toElement, toInclusive));
+ }
+
+ @Override
+ public PCollectionsImmutableNavigableSet<E> headSet(E toElement, boolean inclusive) {
+ return new PCollectionsImmutableNavigableSet<>(underlying.headSet(toElement, inclusive));
+ }
+
+ @Override
+ public PCollectionsImmutableNavigableSet<E> tailSet(E fromElement, boolean inclusive) {
+ return new PCollectionsImmutableNavigableSet<>(underlying.tailSet(fromElement, inclusive));
+ }
+
+ @Override
+ public Comparator<? super E> comparator() {
+ return underlying.comparator();
+ }
+
+ @Override
+ public PCollectionsImmutableNavigableSet<E> subSet(E fromElement, E toElement) {
+ return new PCollectionsImmutableNavigableSet<>(underlying.subSet(fromElement, toElement));
+ }
+
+ @Override
+ public PCollectionsImmutableNavigableSet<E> headSet(E toElement) {
+ return new PCollectionsImmutableNavigableSet<>(underlying.headSet(toElement));
+ }
+
+ @Override
+ public PCollectionsImmutableNavigableSet<E> tailSet(E fromElement) {
+ return new PCollectionsImmutableNavigableSet<>(underlying.tailSet(fromElement));
+ }
+
+ @Override
+ public E first() {
+ return underlying.first();
+ }
+
+ @Override
+ public E last() {
+ return underlying.last();
+ }
+
+ @Override
+ public int size() {
+ return underlying.size();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return underlying.isEmpty();
+ }
+
+ @Override
+ public boolean contains(Object o) {
+ return underlying.contains(o);
+ }
+
+ @Override
+ public Iterator<E> iterator() {
+ return underlying.iterator();
+ }
+
+ @Override
+ public void forEach(Consumer<? super E> action) {
+ underlying.forEach(action);
+ }
+
+ @Override
+ public Object[] toArray() {
+ return underlying.toArray();
+ }
+
+ @Override
+ public <T> T[] toArray(T[] a) {
+ return underlying.toArray(a);
+ }
+
+ @Override
+ public boolean add(E e) {
+ // will throw UnsupportedOperationException
+ return underlying.add(e);
+ }
+
+ @Override
+ public boolean remove(Object o) {
+ // will throw UnsupportedOperationException
+ return underlying.remove(o);
+ }
+
+ @Override
+ public boolean containsAll(Collection<?> c) {
+ return underlying.containsAll(c);
+ }
+
+ @Override
+ public boolean addAll(Collection<? extends E> c) {
+ // will throw UnsupportedOperationException
+ return underlying.addAll(c);
+ }
+
+ @Override
+ public boolean retainAll(Collection<?> c) {
+ // will throw UnsupportedOperationException
+ return underlying.retainAll(c);
+ }
+
+ @Override
+ public boolean removeAll(Collection<?> c) {
+ // will throw UnsupportedOperationException
+ return underlying.removeAll(c);
+ }
+
+ @Override
+ public boolean removeIf(Predicate<? super E> filter) {
+ // will throw UnsupportedOperationException
+ return underlying.removeIf(filter);
+ }
+
+ @Override
+ public void clear() {
+ // will throw UnsupportedOperationException
+ underlying.clear();
+ }
+
+ @Override
+ public Spliterator<E> spliterator() {
+ return underlying().spliterator();
+ }
+
+ @Override
+ public Stream<E> stream() {
+ return underlying().stream();
+ }
+
+ @Override
+ public Stream<E> parallelStream() {
+ return underlying().parallelStream();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ PCollectionsImmutableNavigableSet<?> that = (PCollectionsImmutableNavigableSet<?>) o;
+ return Objects.equals(underlying(), that.underlying());
+ }
+
+ @Override
+ public int hashCode() {
+ return underlying().hashCode();
+ }
+
+ @Override
+ public String toString() {
+ return "PCollectionsImmutableNavigableSet{" +
+ "underlying=" + underlying() +
+ '}';
+ }
+ TreePSet<E> underlying() {
Review Comment:
Add a separator line and the comment `// package-private for testing`
##########
server-common/src/test/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableMapTest.java:
##########
@@ -225,77 +225,59 @@ public void testDelegationOfReplaceAll() {
new PCollectionsHashMapWrapperDelegationChecker<>()
.defineMockConfigurationForVoidMethodInvocation(mock -> mock.replaceAll(eq(mockBiFunction)))
.defineWrapperVoidMethodInvocation(wrapper -> wrapper.replaceAll(mockBiFunction))
- .doVoidMethodDelegationCheck();
+ .doUnsupportedVoidFunctionDelegrationCheck();
}
@Test
public void testDelegationOfPutIfAbsent() {
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.putIfAbsent(eq(this), eq(this)), this)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.putIfAbsent(this, this), identity())
- .doFunctionDelegationCheck();
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testDelegationOfRemoveByKeyAndValue(boolean mockFunctionReturnValue) {
- new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.remove(eq(this), eq(this)), mockFunctionReturnValue)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.remove(this, this), identity())
- .doFunctionDelegationCheck();
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testDelegationOfReplaceWhenMappedToSpecificValue(boolean mockFunctionReturnValue) {
- new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.replace(eq(this), eq(this), eq(this)), mockFunctionReturnValue)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.replace(this, this, this), identity())
- .doFunctionDelegationCheck();
Review Comment:
Looks like we lost these two test methods
##########
server-common/src/main/java/org/apache/kafka/server/immutable/ImmutableNavigableSet.java:
##########
@@ -0,0 +1,59 @@
+/*
+ * 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.server.immutable;
+
+import org.apache.kafka.server.immutable.pcollections.PCollectionsImmutableNavigableSet;
+
+import java.util.NavigableSet;
+
+/**
+ * A persistent Tree-based NavigableSet wrapper
+ * java.util.Set methods that mutate in-place will throw UnsupportedOperationException
+ *
+ * @param <E> the element type
+ */
+public interface ImmutableNavigableSet<E> extends ImmutableSet<E>, NavigableSet<E> {
+ /**
+ * @return a wrapped tree-based persistent navigable set that is empty
+ * @param <E> the element type
+ */
+ static <E extends Comparable<? super E>> ImmutableNavigableSet<E> empty() {
+ return PCollectionsImmutableNavigableSet.empty();
+ }
+
+ /**
+ * @param e the element
+ * @return a wrapped hash-based persistent set that is empty
+ * @param <E> the element type
+ */
+ static <E extends Comparable<? super E>> ImmutableNavigableSet<E> singleton(E e) {
+ return PCollectionsImmutableNavigableSet.singleton(e);
+ }
+
+ /**
+ * @param e the element
+ * @return a wrapped persistent sorted set that differs from this one in that the given element is added (if necessary)
Review Comment:
s/sorted/navigable/
##########
server-common/src/test/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableMapTest.java:
##########
@@ -225,77 +225,59 @@ public void testDelegationOfReplaceAll() {
new PCollectionsHashMapWrapperDelegationChecker<>()
.defineMockConfigurationForVoidMethodInvocation(mock -> mock.replaceAll(eq(mockBiFunction)))
.defineWrapperVoidMethodInvocation(wrapper -> wrapper.replaceAll(mockBiFunction))
- .doVoidMethodDelegationCheck();
+ .doUnsupportedVoidFunctionDelegrationCheck();
}
@Test
public void testDelegationOfPutIfAbsent() {
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.putIfAbsent(eq(this), eq(this)), this)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.putIfAbsent(this, this), identity())
- .doFunctionDelegationCheck();
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testDelegationOfRemoveByKeyAndValue(boolean mockFunctionReturnValue) {
- new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.remove(eq(this), eq(this)), mockFunctionReturnValue)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.remove(this, this), identity())
- .doFunctionDelegationCheck();
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testDelegationOfReplaceWhenMappedToSpecificValue(boolean mockFunctionReturnValue) {
- new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.replace(eq(this), eq(this), eq(this)), mockFunctionReturnValue)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.replace(this, this, this), identity())
- .doFunctionDelegationCheck();
+ .defineMockConfigurationForUnsupportedFunction(mock -> mock.putIfAbsent(eq(this), eq(this)))
+ .defineWrapperUnsupportedFunctionInvocation(wrapper -> wrapper.putIfAbsent(this, this))
+ .doUnsupportedFunctionDelegrationCheck();
}
@Test
public void testDelegationOfReplaceWhenMappedToAnyValue() {
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.replace(eq(this), eq(this)), this)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.replace(this, this), identity())
- .doFunctionDelegationCheck();
+ .defineMockConfigurationForUnsupportedFunction(mock -> mock.replace(eq(this), eq(this)))
+ .defineWrapperUnsupportedFunctionInvocation(wrapper -> wrapper.replace(this, this))
+ .doUnsupportedFunctionDelegrationCheck();
}
@Test
public void testDelegationOfComputeIfAbsent() {
final Function<Object, Object> mockFunction = mock(Function.class);
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForVoidMethodInvocation(mock -> mock.computeIfAbsent(eq(this), eq(mockFunction)))
- .defineWrapperVoidMethodInvocation(wrapper -> wrapper.computeIfAbsent(this, mockFunction))
- .doVoidMethodDelegationCheck();
+ .defineMockConfigurationForUnsupportedFunction(mock -> mock.computeIfAbsent(eq(this), eq(mockFunction)))
+ .defineWrapperUnsupportedFunctionInvocation(wrapper -> wrapper.computeIfAbsent(this, mockFunction))
+ .doUnsupportedFunctionDelegrationCheck();
}
@Test
public void testDelegationOfComputeIfPresent() {
final BiFunction<Object, Object, Object> mockBiFunction = mock(BiFunction.class);
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForVoidMethodInvocation(mock -> mock.computeIfPresent(eq(this), eq(mockBiFunction)))
- .defineWrapperVoidMethodInvocation(wrapper -> wrapper.computeIfPresent(this, mockBiFunction))
- .doVoidMethodDelegationCheck();
+ .defineMockConfigurationForUnsupportedFunction(mock -> mock.computeIfPresent(eq(this), eq(mockBiFunction)))
+ .defineWrapperUnsupportedFunctionInvocation(wrapper -> wrapper.computeIfPresent(this, mockBiFunction))
+ .doUnsupportedFunctionDelegrationCheck();
}
@Test
public void testDelegationOfCompute() {
final BiFunction<Object, Object, Object> mockBiFunction = mock(BiFunction.class);
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForVoidMethodInvocation(mock -> mock.compute(eq(this), eq(mockBiFunction)))
- .defineWrapperVoidMethodInvocation(wrapper -> wrapper.compute(this, mockBiFunction))
- .doVoidMethodDelegationCheck();
+ .defineMockConfigurationForUnsupportedFunction(mock -> mock.compute(eq(this), eq(mockBiFunction)))
+ .defineWrapperUnsupportedFunctionInvocation(wrapper -> wrapper.compute(this, mockBiFunction))
+ .doUnsupportedFunctionDelegrationCheck();
}
@Test
public void testDelegationOfMerge() {
final BiFunction<Object, Object, Object> mockBiFunction = mock(BiFunction.class);
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForVoidMethodInvocation(mock -> mock.merge(eq(this), eq(this), eq(mockBiFunction)))
- .defineWrapperVoidMethodInvocation(wrapper -> wrapper.merge(this, this, mockBiFunction))
- .doVoidMethodDelegationCheck();
+ .defineMockConfigurationForUnsupportedFunction(mock -> mock.merge(eq(this), eq(this), eq(mockBiFunction)))
+ .defineWrapperUnsupportedFunctionInvocation(wrapper -> wrapper.merge(this, this, mockBiFunction))
+ .doUnsupportedFunctionDelegrationCheck();
Review Comment:
Nice catch on fixing this; merge() is a function rather than a void method.
##########
server-common/src/main/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableNavigableSet.java:
##########
@@ -0,0 +1,273 @@
+/*
+ * 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.server.immutable.pcollections;
+
+import org.apache.kafka.server.immutable.ImmutableNavigableSet;
+import org.pcollections.TreePSet;
+
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.Objects;
+import java.util.Spliterator;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.stream.Stream;
+
+@SuppressWarnings("deprecation")
+public class PCollectionsImmutableNavigableSet<E> implements ImmutableNavigableSet<E> {
+ private final TreePSet<E> underlying;
+
+ /**
+ * @return a wrapped tree-based persistent navigable set that is empty
+ * @param <E> the element type
+ */
+ public static <E extends Comparable<? super E>> PCollectionsImmutableNavigableSet<E> empty() {
+ return new PCollectionsImmutableNavigableSet<>(TreePSet.<E>empty());
+ }
+
+ /**
+ * @param e the element
+ * @return a wrapped hash-based persistent set that is empty
+ * @param <E> the element type
+ */
+ public static <E extends Comparable<? super E>> PCollectionsImmutableNavigableSet<E> singleton(E e) {
+ return new PCollectionsImmutableNavigableSet<>(TreePSet.singleton(e));
+ }
+
+ public PCollectionsImmutableNavigableSet(TreePSet<E> underlying) {
+ this.underlying = underlying;
+ }
+
+ @Override
+ public PCollectionsImmutableNavigableSet<E> added(E e) {
+ return new PCollectionsImmutableNavigableSet<>(underlying.plus(e));
+ }
+
+ @Override
+ public PCollectionsImmutableNavigableSet<E> removed(E e) {
+ return new PCollectionsImmutableNavigableSet<>(underlying.minus(e));
+ }
+
+ @Override
+ public E lower(E e) {
+ return underlying.lower(e);
+ }
+
+ @Override
+ public E floor(E e) {
+ return underlying.floor(e);
+ }
+
+ @Override
+ public E ceiling(E e) {
+ return underlying.ceiling(e);
+ }
+
+ @Override
+ public E higher(E e) {
+ return underlying.higher(e);
+ }
+
+ @Override
+ public E pollFirst() {
+ // will throw UnsupportedOperationException
+ return underlying.pollFirst();
+ }
+
+ @Override
+ public E pollLast() {
+ // will throw UnsupportedOperationException
+ return underlying.pollLast();
+ }
+
+ @Override
+ public PCollectionsImmutableNavigableSet<E> descendingSet() {
+ return new PCollectionsImmutableNavigableSet<>(underlying.descendingSet());
+ }
+
+ @Override
+ public Iterator<E> descendingIterator() {
+ return underlying.descendingIterator();
+ }
+
+ @Override
+ public PCollectionsImmutableNavigableSet<E> subSet(E fromElement, boolean fromInclusive, E toElement, boolean toInclusive) {
+ return new PCollectionsImmutableNavigableSet<>(underlying.subSet(fromElement, fromInclusive, toElement, toInclusive));
+ }
+
+ @Override
+ public PCollectionsImmutableNavigableSet<E> headSet(E toElement, boolean inclusive) {
+ return new PCollectionsImmutableNavigableSet<>(underlying.headSet(toElement, inclusive));
+ }
+
+ @Override
+ public PCollectionsImmutableNavigableSet<E> tailSet(E fromElement, boolean inclusive) {
+ return new PCollectionsImmutableNavigableSet<>(underlying.tailSet(fromElement, inclusive));
+ }
+
+ @Override
+ public Comparator<? super E> comparator() {
+ return underlying.comparator();
+ }
+
+ @Override
+ public PCollectionsImmutableNavigableSet<E> subSet(E fromElement, E toElement) {
+ return new PCollectionsImmutableNavigableSet<>(underlying.subSet(fromElement, toElement));
+ }
+
+ @Override
+ public PCollectionsImmutableNavigableSet<E> headSet(E toElement) {
+ return new PCollectionsImmutableNavigableSet<>(underlying.headSet(toElement));
+ }
+
+ @Override
+ public PCollectionsImmutableNavigableSet<E> tailSet(E fromElement) {
+ return new PCollectionsImmutableNavigableSet<>(underlying.tailSet(fromElement));
+ }
+
+ @Override
+ public E first() {
+ return underlying.first();
+ }
+
+ @Override
+ public E last() {
+ return underlying.last();
+ }
+
+ @Override
+ public int size() {
+ return underlying.size();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return underlying.isEmpty();
+ }
+
+ @Override
+ public boolean contains(Object o) {
+ return underlying.contains(o);
+ }
+
+ @Override
+ public Iterator<E> iterator() {
+ return underlying.iterator();
+ }
+
+ @Override
+ public void forEach(Consumer<? super E> action) {
+ underlying.forEach(action);
+ }
+
+ @Override
+ public Object[] toArray() {
+ return underlying.toArray();
+ }
+
+ @Override
+ public <T> T[] toArray(T[] a) {
+ return underlying.toArray(a);
+ }
+
+ @Override
+ public boolean add(E e) {
+ // will throw UnsupportedOperationException
+ return underlying.add(e);
+ }
+
+ @Override
+ public boolean remove(Object o) {
+ // will throw UnsupportedOperationException
+ return underlying.remove(o);
+ }
+
+ @Override
+ public boolean containsAll(Collection<?> c) {
+ return underlying.containsAll(c);
+ }
+
+ @Override
+ public boolean addAll(Collection<? extends E> c) {
+ // will throw UnsupportedOperationException
+ return underlying.addAll(c);
+ }
+
+ @Override
+ public boolean retainAll(Collection<?> c) {
+ // will throw UnsupportedOperationException
+ return underlying.retainAll(c);
+ }
+
+ @Override
+ public boolean removeAll(Collection<?> c) {
+ // will throw UnsupportedOperationException
+ return underlying.removeAll(c);
+ }
+
+ @Override
+ public boolean removeIf(Predicate<? super E> filter) {
+ // will throw UnsupportedOperationException
+ return underlying.removeIf(filter);
+ }
+
+ @Override
+ public void clear() {
+ // will throw UnsupportedOperationException
+ underlying.clear();
Review Comment:
It looks like I had a mix of invoking `underlying.method()` vs. `underlying().method()` in `PCollections.ImmutableMap`. Can you make everything consistently use `underlying().method()` both there and here? `PCollections.ImmutableSet` seems fine; it's just the map class and here that are now using an inconsistent style. Let's make them all use the same style. Apologies for my oversight there, which seems to have crept in here.
##########
server-common/src/main/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableNavigableSet.java:
##########
@@ -0,0 +1,273 @@
+/*
+ * 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.server.immutable.pcollections;
+
+import org.apache.kafka.server.immutable.ImmutableNavigableSet;
+import org.pcollections.TreePSet;
+
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.Objects;
+import java.util.Spliterator;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+import java.util.stream.Stream;
+
+@SuppressWarnings("deprecation")
+public class PCollectionsImmutableNavigableSet<E> implements ImmutableNavigableSet<E> {
+ private final TreePSet<E> underlying;
+
+ /**
+ * @return a wrapped tree-based persistent navigable set that is empty
+ * @param <E> the element type
+ */
+ public static <E extends Comparable<? super E>> PCollectionsImmutableNavigableSet<E> empty() {
+ return new PCollectionsImmutableNavigableSet<>(TreePSet.<E>empty());
+ }
+
+ /**
+ * @param e the element
+ * @return a wrapped hash-based persistent set that is empty
Review Comment:
Looks like copy/paste error -- not hash-based.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] emissionnebula commented on pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "emissionnebula (via GitHub)" <gi...@apache.org>.
emissionnebula commented on PR #13437:
URL: https://github.com/apache/kafka/pull/13437#issuecomment-1513618207
A lot of [tests](https://ci-builds.apache.org/blue/rest/organizations/jenkins/pipelines/Kafka/pipelines/kafka-pr/branches/PR-13437/runs/8/log/?start=0) related to Connect seems to be failing. But those are unrelated. I could see a similar set of tests failing in [other PRs](https://ci-builds.apache.org/blue/rest/organizations/jenkins/pipelines/Kafka/pipelines/kafka-pr/branches/PR-13595/runs/3/log/?start=0) as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] emissionnebula commented on pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "emissionnebula (via GitHub)" <gi...@apache.org>.
emissionnebula commented on PR #13437:
URL: https://github.com/apache/kafka/pull/13437#issuecomment-1512536862
I have updated the PR following the conclusion of the discussion on https://github.com/apache/kafka/pull/13280 and have added a wrapper class with an interface for NavigableSet to use with Authorizer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] omkreddy merged pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "omkreddy (via GitHub)" <gi...@apache.org>.
omkreddy merged PR #13437:
URL: https://github.com/apache/kafka/pull/13437
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] divijvaidya commented on pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on PR #13437:
URL: https://github.com/apache/kafka/pull/13437#issuecomment-1488637379
Very interesting change! We can follow this pattern in many other places in the code base.
However, I would like to understand first why we didn't consider usage of native Java's data structures such as CopyOnWrite* data structures. Also, please note that the usage of Immutable data structures is nice for cases with low write and high read throughput use cases. It would be nice if you could mention why the trade-off with immutable data structures is worth it in the case of AclAuthorizer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on PR #13437:
URL: https://github.com/apache/kafka/pull/13437#issuecomment-1496504250
> We can probably minimize the risk of changing the code at all usage points by defining interfaces and wrapper classes in our code to standardize the usage of persistent collections. In the future, if we decide to move away from one library to another then it wouldn't require changing the code everywhere.
Yes, I added such wrappers in https://github.com/apache/kafka/pull/13280/. They just cover unordered sets/maps, and they have not been reviewed yet, but we would want wrappers for ordered sets/maps as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] emissionnebula commented on a diff in pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "emissionnebula (via GitHub)" <gi...@apache.org>.
emissionnebula commented on code in PR #13437:
URL: https://github.com/apache/kafka/pull/13437#discussion_r1169727051
##########
jmh-benchmarks/src/main/java/org/apache/kafka/jmh/acl/AuthorizerUpdateBenchmark.java:
##########
@@ -0,0 +1,114 @@
+package org.apache.kafka.jmh.acl;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.resource.PatternType;
+import org.apache.kafka.common.resource.ResourcePattern;
+import org.apache.kafka.common.resource.ResourceType;
+import org.apache.kafka.common.security.auth.KafkaPrincipal;
+import org.apache.kafka.metadata.authorizer.StandardAcl;
+import org.apache.kafka.metadata.authorizer.StandardAclWithId;
+import org.apache.kafka.metadata.authorizer.StandardAuthorizer;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.kafka.common.acl.AclOperation.READ;
+import static org.apache.kafka.common.acl.AclPermissionType.ALLOW;
+
+@State(Scope.Benchmark)
+@Fork(value = 1)
+@Warmup(iterations = 0)
+@Measurement(iterations = 4)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+public class AuthorizerUpdateBenchmark {
Review Comment:
Renamed to StandardAuthorizerUpdateBenchmark
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] omkreddy commented on a diff in pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "omkreddy (via GitHub)" <gi...@apache.org>.
omkreddy commented on code in PR #13437:
URL: https://github.com/apache/kafka/pull/13437#discussion_r1169638781
##########
jmh-benchmarks/src/main/java/org/apache/kafka/jmh/acl/AuthorizerUpdateBenchmark.java:
##########
@@ -0,0 +1,114 @@
+package org.apache.kafka.jmh.acl;
Review Comment:
nit: missing license
##########
jmh-benchmarks/src/main/java/org/apache/kafka/jmh/acl/AuthorizerUpdateBenchmark.java:
##########
@@ -0,0 +1,114 @@
+package org.apache.kafka.jmh.acl;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.resource.PatternType;
+import org.apache.kafka.common.resource.ResourcePattern;
+import org.apache.kafka.common.resource.ResourceType;
+import org.apache.kafka.common.security.auth.KafkaPrincipal;
+import org.apache.kafka.metadata.authorizer.StandardAcl;
+import org.apache.kafka.metadata.authorizer.StandardAclWithId;
+import org.apache.kafka.metadata.authorizer.StandardAuthorizer;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.kafka.common.acl.AclOperation.READ;
+import static org.apache.kafka.common.acl.AclPermissionType.ALLOW;
+
+@State(Scope.Benchmark)
+@Fork(value = 1)
+@Warmup(iterations = 0)
+@Measurement(iterations = 4)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+public class AuthorizerUpdateBenchmark {
Review Comment:
Can we rename to `AuthorizerUpdateBenchmark `
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] emissionnebula commented on pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "emissionnebula (via GitHub)" <gi...@apache.org>.
emissionnebula commented on PR #13437:
URL: https://github.com/apache/kafka/pull/13437#issuecomment-1485100563
Oops. Thanks @rondagostino for pointing this out. I have updated the description with the correct details.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] omkreddy commented on pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "omkreddy (via GitHub)" <gi...@apache.org>.
omkreddy commented on PR #13437:
URL: https://github.com/apache/kafka/pull/13437#issuecomment-1517476396
Test failures are known connect/mirror failures: https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka-pr/detail/PR-13437/12/tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] Hangleton commented on a diff in pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "Hangleton (via GitHub)" <gi...@apache.org>.
Hangleton commented on code in PR #13437:
URL: https://github.com/apache/kafka/pull/13437#discussion_r1173645096
##########
metadata/src/main/java/org/apache/kafka/metadata/authorizer/AclCache.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.metadata.authorizer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.acl.AclBinding;
+import org.apache.kafka.common.acl.AclBindingFilter;
+import org.apache.kafka.server.immutable.ImmutableMap;
+import org.apache.kafka.server.immutable.ImmutableNavigableSet;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * An immutable class that stores the ACLs in KRaft-based clusters.
+ */
+public class AclCache {
+ /**
+ * Contains all of the current ACLs sorted by (resource type, resource name).
+ */
+ private final ImmutableNavigableSet<StandardAcl> aclsByResource;
+
+ /**
+ * Contains all of the current ACLs indexed by UUID.
+ */
+ private final ImmutableMap<Uuid, StandardAcl> aclsById;
+
+ AclCache() {
+ this(ImmutableNavigableSet.empty(), ImmutableMap.empty());
+ }
+
+ private AclCache(final ImmutableNavigableSet<StandardAcl> aclsByResource, final ImmutableMap<Uuid, StandardAcl> aclsById) {
+ this.aclsByResource = aclsByResource;
+ this.aclsById = aclsById;
+ }
+
+ public ImmutableNavigableSet<StandardAcl> aclsByResource() {
+ return aclsByResource;
+ }
+
+ Iterable<AclBinding> acls(AclBindingFilter filter) {
+ List<AclBinding> aclBindingList = new ArrayList<>();
+ aclsByResource.forEach(acl -> {
+ AclBinding aclBinding = acl.toBinding();
+ if (filter.matches(aclBinding)) {
+ aclBindingList.add(aclBinding);
+ }
+ });
+ return aclBindingList;
+ }
+
+ int count() {
+ return aclsById.size();
+ }
+
+ StandardAcl getAcl(Uuid id) {
+ return aclsById.get(id);
+ }
+
+ AclCache addAcl(Uuid id, StandardAcl acl) {
Review Comment:
> *Since writes are done on a single thread, the only case of concurrency we have to solve here is when multiple reads and a single write are happening in parallel.*
Do I get this right that the single writer assumption stated in the PR description is critical to achieve consistency in the sequence of operations below? (e.g. that the state checked line 77 is still valid line 81). Should multiple writes happen concurrently, this would not be the case, right? Is there a way to enforce the single writer condition? Or, shouldn't the cache preserve consistency under multiple writers (since it has no control over how many actors can update its state concurrently)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] emissionnebula commented on a diff in pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "emissionnebula (via GitHub)" <gi...@apache.org>.
emissionnebula commented on code in PR #13437:
URL: https://github.com/apache/kafka/pull/13437#discussion_r1172848492
##########
server-common/src/test/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableMapTest.java:
##########
@@ -225,77 +225,59 @@ public void testDelegationOfReplaceAll() {
new PCollectionsHashMapWrapperDelegationChecker<>()
.defineMockConfigurationForVoidMethodInvocation(mock -> mock.replaceAll(eq(mockBiFunction)))
.defineWrapperVoidMethodInvocation(wrapper -> wrapper.replaceAll(mockBiFunction))
- .doVoidMethodDelegationCheck();
+ .doUnsupportedVoidFunctionDelegrationCheck();
}
@Test
public void testDelegationOfPutIfAbsent() {
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.putIfAbsent(eq(this), eq(this)), this)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.putIfAbsent(this, this), identity())
- .doFunctionDelegationCheck();
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testDelegationOfRemoveByKeyAndValue(boolean mockFunctionReturnValue) {
- new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.remove(eq(this), eq(this)), mockFunctionReturnValue)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.remove(this, this), identity())
- .doFunctionDelegationCheck();
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testDelegationOfReplaceWhenMappedToSpecificValue(boolean mockFunctionReturnValue) {
- new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.replace(eq(this), eq(this), eq(this)), mockFunctionReturnValue)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.replace(this, this, this), identity())
- .doFunctionDelegationCheck();
Review Comment:
These two functions are not overridden by PCollection classes. `mock.replace()` or `mock.remove()` will call Java's default implementation in `Map.replace()` or `AbstractMap.remove()`. Mocking that testing that seemed not required.
For example: `Map.replace()` looks like this:
```
default boolean replace(K key, V oldValue, V newValue) {
Object curValue = get(key);
if (!Objects.equals(curValue, oldValue) ||
(curValue == null && !containsKey(key))) {
return false;
}
put(key, newValue);
return true;
}
```
Since it internally calls `put()` which we are already testing, so that's why I thought it was not necessary to test these functions. Likewise for `remove()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on PR #13437:
URL: https://github.com/apache/kafka/pull/13437#issuecomment-1483915474
Thanks @emissionnebula! This is great information to consider when evaluating these libraries. A requirement to use `tailSet()` on a sorted set does not exist in the metadata cache, so the poor performance of this function in Paguro and the lack of functionality in Vavr did not arise there. It is very helpful to know this as we decide how to proceed.
One question about the `TLDR` section. It says for the read performance of PCollections:
[PCollections] Read Performance | ❌ (10000x of AclAuthorizer)
That seems to be an incorrect summarization -- is it indeed a mistake?
I'll update https://github.com/apache/kafka/pull/13280 to use PCollections and see what the benchmarks indicate.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] emissionnebula commented on pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "emissionnebula (via GitHub)" <gi...@apache.org>.
emissionnebula commented on PR #13437:
URL: https://github.com/apache/kafka/pull/13437#issuecomment-1493821132
> Since you mentioned, "authorizer is quite read-heavy with very infrequent writes", and given the above risk highlighted above, would having slow writes with CopyOnWrite be an acceptable trade-off?
There is another [PR](https://github.com/apache/kafka/pull/13280) by @rondagostino that evaluates these libraries for the Kraft metadata image change performance. That code currently uses the CopyOnWrite approach. The performance of these persistent libraries over existing code is multi-magnitude better - CopyOnWrite = ~31sec/op and PCollection = 509 ns/op.
> let's say we end up using a dependency that is maintained by a single person and they decide to stop maintaining it.
I totally agree with this risk but the gain with these libraries is quite significant so maybe we can consider this also as a trade-off.
> That leaves us with a choice to either overhaul our code base and remove all usages of that library (which I expect to be very intrusive for locking data structures) or take over ownership of something that we as a community are not experts in.
We can probably minimize the risk of changing the code at all usage points by defining interfaces and wrapper classes in our code to standardize the usage of persistent collections. In the future, if we decide to move away from one library to another then it wouldn't require changing the code everywhere.
cc - @ijuma @mimaison @cmccabe @omkreddy @dajac, what are your views on this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] emissionnebula commented on a diff in pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "emissionnebula (via GitHub)" <gi...@apache.org>.
emissionnebula commented on code in PR #13437:
URL: https://github.com/apache/kafka/pull/13437#discussion_r1172850344
##########
server-common/src/test/java/org/apache/kafka/server/immutable/DelegationChecker.java:
##########
@@ -139,6 +163,16 @@ public void doFunctionDelegationCheck() {
}
}
+ public void doUnsupportedFunctionDelegrationCheck() {
Review Comment:
yup, I thought it made sense to test it explicitly.
Updated the functions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] emissionnebula commented on a diff in pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "emissionnebula (via GitHub)" <gi...@apache.org>.
emissionnebula commented on code in PR #13437:
URL: https://github.com/apache/kafka/pull/13437#discussion_r1172848492
##########
server-common/src/test/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableMapTest.java:
##########
@@ -225,77 +225,59 @@ public void testDelegationOfReplaceAll() {
new PCollectionsHashMapWrapperDelegationChecker<>()
.defineMockConfigurationForVoidMethodInvocation(mock -> mock.replaceAll(eq(mockBiFunction)))
.defineWrapperVoidMethodInvocation(wrapper -> wrapper.replaceAll(mockBiFunction))
- .doVoidMethodDelegationCheck();
+ .doUnsupportedVoidFunctionDelegrationCheck();
}
@Test
public void testDelegationOfPutIfAbsent() {
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.putIfAbsent(eq(this), eq(this)), this)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.putIfAbsent(this, this), identity())
- .doFunctionDelegationCheck();
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testDelegationOfRemoveByKeyAndValue(boolean mockFunctionReturnValue) {
- new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.remove(eq(this), eq(this)), mockFunctionReturnValue)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.remove(this, this), identity())
- .doFunctionDelegationCheck();
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testDelegationOfReplaceWhenMappedToSpecificValue(boolean mockFunctionReturnValue) {
- new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.replace(eq(this), eq(this), eq(this)), mockFunctionReturnValue)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.replace(this, this, this), identity())
- .doFunctionDelegationCheck();
Review Comment:
These two functions are not overridden by PCollection classes. `mock.replace()` or `mock.remove()` will call Java's default implementation in `Map.replace()` or `AbstractMap.remove()`. Mocking and testing that I felt is not required.
For example: `Map.replace()` looks like this:
```
default boolean replace(K key, V oldValue, V newValue) {
Object curValue = get(key);
if (!Objects.equals(curValue, oldValue) ||
(curValue == null && !containsKey(key))) {
return false;
}
put(key, newValue);
return true;
}
```
Since it internally calls `put()` which we are already testing, so that's why I thought it was not necessary to test these functions. Likewise for `remove()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] emissionnebula commented on pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "emissionnebula (via GitHub)" <gi...@apache.org>.
emissionnebula commented on PR #13437:
URL: https://github.com/apache/kafka/pull/13437#issuecomment-1512728910
@rondagostino - I have made changes to the immutable class tests for the methods that are not supported due to immutability such as `add`, `put`, etc. Earlier we were mocking the method of the underlying object even in case of add/put to return some value but ideally, those functions should be returning `UnsupportedOperationException`. I have changed those tests to check for the exception.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] divijvaidya commented on pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on PR #13437:
URL: https://github.com/apache/kafka/pull/13437#issuecomment-1490628418
> CopyOnWrite is quite inefficient compared to persistent collections
I agree on this that persistent collections are more performant than standard Java `CopyOnWrite`.
But we also, need to consider the impact of taking a dependency over a new project. Hypothetically (I haven't looked at the community of the two projects that you propose), let's say we end up using a dependency which is maintained by a single person and they decide to stop maintaining it. That leaves us with a choice to either overhaul our code base and remove all usages of that library (which I expect to be very intrusive for locking data structures) or take over ownership of something that we as a community are not experts in. To de-risk this, we would probably prefer libraries which have a strong community (such as Apache projects) or have prevalent in majority of the modern architecture such that the scenario of development contonuity stops being a risk (e.g. netty).
Since you mentioned, "authorizer is quite read-heavy with very infrequent writes", and given the above risk highlighted above, would having slow writes with `CopyOnWrite` be an acceptable trade-off?
I am curious to hear what others in the community have to say about this.
cc: @ijuma @dajac @mimaison, perhaps you folks might be interested in this discussion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] emissionnebula commented on pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "emissionnebula (via GitHub)" <gi...@apache.org>.
emissionnebula commented on PR #13437:
URL: https://github.com/apache/kafka/pull/13437#issuecomment-1490312456
Thanks @divijvaidya for the comment.
> why we didn't consider usage of native Java's data structures such as CopyOnWrite* data structures.
CopyOnWrite is quite inefficient compared to persistent collections since it requires copying the whole collections for each update. I have given an example in the description which explains why persistent collections are efficient.
> It would be nice if you could mention why the trade-off with immutable data structures is worth it in the case of AclAuthorizer.
Added in the description.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] emissionnebula commented on a diff in pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "emissionnebula (via GitHub)" <gi...@apache.org>.
emissionnebula commented on code in PR #13437:
URL: https://github.com/apache/kafka/pull/13437#discussion_r1177916880
##########
metadata/src/main/java/org/apache/kafka/metadata/authorizer/AclCache.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.metadata.authorizer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.acl.AclBinding;
+import org.apache.kafka.common.acl.AclBindingFilter;
+import org.apache.kafka.server.immutable.ImmutableMap;
+import org.apache.kafka.server.immutable.ImmutableNavigableSet;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * An immutable class that stores the ACLs in KRaft-based clusters.
+ */
+public class AclCache {
+ /**
+ * Contains all of the current ACLs sorted by (resource type, resource name).
+ */
+ private final ImmutableNavigableSet<StandardAcl> aclsByResource;
+
+ /**
+ * Contains all of the current ACLs indexed by UUID.
+ */
+ private final ImmutableMap<Uuid, StandardAcl> aclsById;
+
+ AclCache() {
+ this(ImmutableNavigableSet.empty(), ImmutableMap.empty());
+ }
+
+ private AclCache(final ImmutableNavigableSet<StandardAcl> aclsByResource, final ImmutableMap<Uuid, StandardAcl> aclsById) {
+ this.aclsByResource = aclsByResource;
+ this.aclsById = aclsById;
+ }
+
+ public ImmutableNavigableSet<StandardAcl> aclsByResource() {
+ return aclsByResource;
+ }
+
+ Iterable<AclBinding> acls(AclBindingFilter filter) {
+ List<AclBinding> aclBindingList = new ArrayList<>();
+ aclsByResource.forEach(acl -> {
+ AclBinding aclBinding = acl.toBinding();
+ if (filter.matches(aclBinding)) {
+ aclBindingList.add(aclBinding);
+ }
+ });
+ return aclBindingList;
+ }
+
+ int count() {
+ return aclsById.size();
+ }
+
+ StandardAcl getAcl(Uuid id) {
+ return aclsById.get(id);
+ }
+
+ AclCache addAcl(Uuid id, StandardAcl acl) {
Review Comment:
> > _Since writes are done on a single thread, the only case of concurrency we have to solve here is when multiple reads and a single write are happening in parallel._
>
> Do I get this right that the single writer assumption stated in the PR description is critical to achieve consistency in the sequence of operations below? (e.g. that the state checked line 77 is still valid line 81). Should multiple writes happen concurrently, this would not be the case, right? Is there a way to enforce the single writer condition? Or, shouldn't the cache preserve consistency under multiple writers (since it has no control over how many actors can update its state concurrently)?
Thanks @Hangleton for the comment. This condition of single write will always be true for Authorizer because we have to apply the ACL changes in the order of their arrival. In case of Kraft, that order will be the order in which it is written to metadata topic. So we would never enable multiple threads to read from the metadata topic and write to AclCache. Due to this I didn't add a lock on writes here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a diff in pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on code in PR #13437:
URL: https://github.com/apache/kafka/pull/13437#discussion_r1173187845
##########
server-common/src/test/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableMapTest.java:
##########
@@ -225,77 +225,59 @@ public void testDelegationOfReplaceAll() {
new PCollectionsHashMapWrapperDelegationChecker<>()
.defineMockConfigurationForVoidMethodInvocation(mock -> mock.replaceAll(eq(mockBiFunction)))
.defineWrapperVoidMethodInvocation(wrapper -> wrapper.replaceAll(mockBiFunction))
- .doVoidMethodDelegationCheck();
+ .doUnsupportedVoidFunctionDelegrationCheck();
}
@Test
public void testDelegationOfPutIfAbsent() {
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.putIfAbsent(eq(this), eq(this)), this)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.putIfAbsent(this, this), identity())
- .doFunctionDelegationCheck();
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testDelegationOfRemoveByKeyAndValue(boolean mockFunctionReturnValue) {
- new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.remove(eq(this), eq(this)), mockFunctionReturnValue)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.remove(this, this), identity())
- .doFunctionDelegationCheck();
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testDelegationOfReplaceWhenMappedToSpecificValue(boolean mockFunctionReturnValue) {
- new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.replace(eq(this), eq(this), eq(this)), mockFunctionReturnValue)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.replace(this, this, this), identity())
- .doFunctionDelegationCheck();
Review Comment:
Yeah, this is why I said in the comment above:
```
I think on the one hand we don't have to test that the real method throws the exception -- that's up to the library itself to test in its test suite. All we have to test is that we delegate correctly.
```
The underlying library implementation in I guess these two cases doesn't explicitly throw the exception but instead reuses a default implementation from `java.util`, and that will cause an exception to be thrown if the map ends up needing to be mutated.
At a minimum we should put the two test cases back in as I wrote them. If you want to do it with an explicit check for the thrown exception by invoking the underlying method then you may have to do a bit more configuration on the mock so that the underlying method will actually decide it needs to mutate the map. probably just putting my stuff back in is easier (and also add "UnsupportedOperation" into the test method names).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] emissionnebula commented on a diff in pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "emissionnebula (via GitHub)" <gi...@apache.org>.
emissionnebula commented on code in PR #13437:
URL: https://github.com/apache/kafka/pull/13437#discussion_r1173353923
##########
server-common/src/test/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableMapTest.java:
##########
@@ -68,7 +68,7 @@ public void testUnderlying() {
}
@Test
- public void testDelegationOfAfterAdding() {
+ public void testDelegationOfAfterUpdated() {
Review Comment:
ohh. I missed this :(
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] omkreddy commented on a diff in pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "omkreddy (via GitHub)" <gi...@apache.org>.
omkreddy commented on code in PR #13437:
URL: https://github.com/apache/kafka/pull/13437#discussion_r1169692194
##########
jmh-benchmarks/src/main/java/org/apache/kafka/jmh/acl/AuthorizerUpdateBenchmark.java:
##########
@@ -0,0 +1,114 @@
+package org.apache.kafka.jmh.acl;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.resource.PatternType;
+import org.apache.kafka.common.resource.ResourcePattern;
+import org.apache.kafka.common.resource.ResourceType;
+import org.apache.kafka.common.security.auth.KafkaPrincipal;
+import org.apache.kafka.metadata.authorizer.StandardAcl;
+import org.apache.kafka.metadata.authorizer.StandardAclWithId;
+import org.apache.kafka.metadata.authorizer.StandardAuthorizer;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.kafka.common.acl.AclOperation.READ;
+import static org.apache.kafka.common.acl.AclPermissionType.ALLOW;
+
+@State(Scope.Benchmark)
+@Fork(value = 1)
+@Warmup(iterations = 0)
+@Measurement(iterations = 4)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+public class AuthorizerUpdateBenchmark {
Review Comment:
Can we rename to `StandardAuthorizerUpdateBenchmark `
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] Hangleton commented on a diff in pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "Hangleton (via GitHub)" <gi...@apache.org>.
Hangleton commented on code in PR #13437:
URL: https://github.com/apache/kafka/pull/13437#discussion_r1173645096
##########
metadata/src/main/java/org/apache/kafka/metadata/authorizer/AclCache.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.metadata.authorizer;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.acl.AclBinding;
+import org.apache.kafka.common.acl.AclBindingFilter;
+import org.apache.kafka.server.immutable.ImmutableMap;
+import org.apache.kafka.server.immutable.ImmutableNavigableSet;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * An immutable class that stores the ACLs in KRaft-based clusters.
+ */
+public class AclCache {
+ /**
+ * Contains all of the current ACLs sorted by (resource type, resource name).
+ */
+ private final ImmutableNavigableSet<StandardAcl> aclsByResource;
+
+ /**
+ * Contains all of the current ACLs indexed by UUID.
+ */
+ private final ImmutableMap<Uuid, StandardAcl> aclsById;
+
+ AclCache() {
+ this(ImmutableNavigableSet.empty(), ImmutableMap.empty());
+ }
+
+ private AclCache(final ImmutableNavigableSet<StandardAcl> aclsByResource, final ImmutableMap<Uuid, StandardAcl> aclsById) {
+ this.aclsByResource = aclsByResource;
+ this.aclsById = aclsById;
+ }
+
+ public ImmutableNavigableSet<StandardAcl> aclsByResource() {
+ return aclsByResource;
+ }
+
+ Iterable<AclBinding> acls(AclBindingFilter filter) {
+ List<AclBinding> aclBindingList = new ArrayList<>();
+ aclsByResource.forEach(acl -> {
+ AclBinding aclBinding = acl.toBinding();
+ if (filter.matches(aclBinding)) {
+ aclBindingList.add(aclBinding);
+ }
+ });
+ return aclBindingList;
+ }
+
+ int count() {
+ return aclsById.size();
+ }
+
+ StandardAcl getAcl(Uuid id) {
+ return aclsById.get(id);
+ }
+
+ AclCache addAcl(Uuid id, StandardAcl acl) {
Review Comment:
> *Since writes are done on a single thread, the only case of concurrency we have to solve here is when multiple reads and a single write are happening in parallel.*
Do I get this right that the single writer assumption stated in the PR description is critical to achieve consistency in the sequence of operations below? (e.g. that the state checked line 77 is still valid line 81). Should multiple writes happen concurrently, this would not be the case, right? Is there a way to enforce it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] emissionnebula commented on a diff in pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "emissionnebula (via GitHub)" <gi...@apache.org>.
emissionnebula commented on code in PR #13437:
URL: https://github.com/apache/kafka/pull/13437#discussion_r1172848492
##########
server-common/src/test/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableMapTest.java:
##########
@@ -225,77 +225,59 @@ public void testDelegationOfReplaceAll() {
new PCollectionsHashMapWrapperDelegationChecker<>()
.defineMockConfigurationForVoidMethodInvocation(mock -> mock.replaceAll(eq(mockBiFunction)))
.defineWrapperVoidMethodInvocation(wrapper -> wrapper.replaceAll(mockBiFunction))
- .doVoidMethodDelegationCheck();
+ .doUnsupportedVoidFunctionDelegrationCheck();
}
@Test
public void testDelegationOfPutIfAbsent() {
new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.putIfAbsent(eq(this), eq(this)), this)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.putIfAbsent(this, this), identity())
- .doFunctionDelegationCheck();
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testDelegationOfRemoveByKeyAndValue(boolean mockFunctionReturnValue) {
- new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.remove(eq(this), eq(this)), mockFunctionReturnValue)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.remove(this, this), identity())
- .doFunctionDelegationCheck();
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testDelegationOfReplaceWhenMappedToSpecificValue(boolean mockFunctionReturnValue) {
- new PCollectionsHashMapWrapperDelegationChecker<>()
- .defineMockConfigurationForFunctionInvocation(mock -> mock.replace(eq(this), eq(this), eq(this)), mockFunctionReturnValue)
- .defineWrapperFunctionInvocationAndMockReturnValueTransformation(wrapper -> wrapper.replace(this, this, this), identity())
- .doFunctionDelegationCheck();
Review Comment:
These two functions are not overridden by PCollection classes. `mock.replace()` or `mock.remove()` will call Java's default implementation in `Map.replace()` or `AbstractMap.remove()`. Mocking that testing that seemed not required.
For example: `Map.replace()` looks like this:
```
default boolean replace(K key, V oldValue, V newValue) {
Object curValue = get(key);
if (!Objects.equals(curValue, oldValue) ||
(curValue == null && !containsKey(key))) {
return false;
}
put(key, newValue);
return true;
}
```
Since it internally calls put which we are already testing, so that's why I thought it was not necessary to test these functions. Likewise for `remove()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a diff in pull request #13437: KAFKA-14828: Remove R/W locks using persistent data structures
Posted by "rondagostino (via GitHub)" <gi...@apache.org>.
rondagostino commented on code in PR #13437:
URL: https://github.com/apache/kafka/pull/13437#discussion_r1173188783
##########
server-common/src/test/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableMapTest.java:
##########
@@ -77,7 +77,7 @@ public void testDelegationOfAfterAdding() {
}
@Test
- public void testDelegationOfAfterRemoving() {
+ public void testDelegationOfAfterRemoved() {
Review Comment:
s/AfterRemoved/Removed/
##########
server-common/src/test/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableSetTest.java:
##########
@@ -71,7 +71,7 @@ public void testUnderlying() {
}
@Test
- public void testDelegationOfAfterAdding() {
+ public void testDelegationOfAfterAdded() {
Review Comment:
s/AfterAdded/Added/
##########
server-common/src/test/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableNavigableSetTest.java:
##########
@@ -82,7 +82,7 @@ public void testDelegationOfAfterAdding() {
}
@Test
- public void testDelegationOfAfterRemoving() {
+ public void testDelegationOfAfterRemoved() {
Review Comment:
s/AfterRemoved/Removed/
##########
server-common/src/test/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableSetTest.java:
##########
@@ -80,7 +80,7 @@ public void testDelegationOfAfterAdding() {
}
@Test
- public void testDelegationOfAfterRemoving() {
+ public void testDelegationOfAfterRemoved() {
Review Comment:
s/AfterRemoved/Removed/
##########
server-common/src/test/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableNavigableSetTest.java:
##########
@@ -73,7 +73,7 @@ public void testUnderlying() {
}
@Test
- public void testDelegationOfAfterAdding() {
+ public void testDelegationOfAfterAdded() {
Review Comment:
s/AfterAdded/Added/
##########
server-common/src/test/java/org/apache/kafka/server/immutable/pcollections/PCollectionsImmutableMapTest.java:
##########
@@ -68,7 +68,7 @@ public void testUnderlying() {
}
@Test
- public void testDelegationOfAfterAdding() {
+ public void testDelegationOfAfterUpdated() {
Review Comment:
s/AfterUpdated/Updated/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org