Merge remote-tracking branch 'origin/kip-1170-format' into kip-1170-format

Author: mannoopj

build.gradle

@@ -1004,6 +1004,8 @@ project(':server') {
     archivesName = "kafka-server"
   }
 
+  apply plugin: 'java-test-fixtures'
+
   dependencies {
     compileOnly libs.bndlib
     compileOnly libs.spotbugs
@@ -1033,6 +1035,13 @@ project(':server') {
     testImplementation project(':test-common:test-common-runtime')
     testImplementation testFixtures(project(':storage:storage-api'))
     testImplementation testFixtures(project(':server-common'))
+    testImplementation testFixtures(project(':metadata'))
+
+    testFixturesImplementation project(':clients')
+    testFixturesImplementation testFixtures(project(':clients'))
+    testFixturesImplementation project(':metadata')
+    testFixturesImplementation project(':server-common')
+    testFixturesImplementation libs.junitJupiter
 
     testRuntimeOnly runtimeTestLibs
   }
@@ -1152,7 +1161,7 @@ project(':core') {
     testImplementation testFixtures(project(':raft'))
     testImplementation testFixtures(project(':server-common'))
     testImplementation testFixtures(project(':storage:storage-api'))
-    testImplementation project(':server').sourceSets.test.output
+    testImplementation testFixtures(project(':server'))
     testImplementation project(':streams')
     testImplementation project(':test-common:test-common-runtime')
     testImplementation project(':test-common:test-common-internal-api')
@@ -2430,6 +2439,8 @@ project(':storage') {
     archivesName = "kafka-storage"
   }
 
+  apply plugin: 'java-test-fixtures'
+
   configurations {
     generator
   }
@@ -2446,17 +2457,25 @@ project(':storage') {
     implementation libs.jacksonDatabind
     implementation libs.metrics
 
+    testFixturesImplementation project(':storage:storage-api')
+    testFixturesImplementation project(':clients')
+    testFixturesImplementation testFixtures(project(':clients'))
+    testFixturesImplementation project(':server-common')
+    testFixturesImplementation libs.slf4jApi
+
     testImplementation project(':clients')
     testImplementation testFixtures(project(':clients'))
     testImplementation project(':core')
     testImplementation project(':core').sourceSets.test.output
     testImplementation testFixtures(project(':storage:storage-api'))
+    testImplementation project(':metadata')
     testImplementation project(':test-common:test-common-internal-api')
     testImplementation project(':test-common:test-common-runtime')
     testImplementation project(':test-common:test-common-util')
     testImplementation project(':server')
     testImplementation project(':server-common')
     testImplementation testFixtures(project(':server-common'))
+    testImplementation testFixtures(project(':metadata'))
     testImplementation project(':transaction-coordinator')
     testImplementation libs.hamcrest
     testImplementation libs.jacksonDataformatYaml
@@ -2608,6 +2627,12 @@ project(':tools') {
 
   configurations {
     releaseOnly
+    // ApacheDS 2.0.0-M24 pulls in the stale bcprov-jdk15on:1.56, which ships the same
+    // RosstandartObjectIdentifiers class as the modern bcprov-jdk18on we already depend on
+    // but is missing fields referenced by bcpkix-jdk18on:1.84. When IntelliJ's test runner
+    // orders the old JAR first, BC provider registration fails with NoSuchFieldError. Drop it.
+    testCompileClasspath.exclude group: 'org.bouncycastle', module: 'bcprov-jdk15on'
+    testRuntimeClasspath.exclude group: 'org.bouncycastle', module: 'bcprov-jdk15on'
   }
 
   dependencies {
@@ -2642,8 +2667,8 @@ project(':tools') {
     testImplementation project(':clients')
     testImplementation testFixtures(project(':clients'))
     testImplementation project(':server')
-    testImplementation project(':server').sourceSets.test.output
     testImplementation project(':core')
+    testImplementation testFixtures(project(':server'))
     testImplementation project(':core').sourceSets.test.output
     testImplementation project(':test-common:test-common-internal-api')
     testImplementation project(':test-common:test-common-runtime')
@@ -2653,7 +2678,7 @@ project(':tools') {
     testImplementation project(':connect:runtime')
     testImplementation project(':connect:runtime').sourceSets.test.output
     testImplementation project(':storage:storage-api').sourceSets.main.output
-    testImplementation project(':storage').sourceSets.test.output
+    testImplementation testFixtures(project(':storage'))
     testImplementation project(':streams')
     testImplementation project(':streams').sourceSets.test.output
     testImplementation project(':streams:integration-tests').sourceSets.test.output
@@ -3552,7 +3577,6 @@ project(':jmh-benchmarks') {
     implementation testFixtures(project(':clients'))
     implementation testFixtures(project(':server-common'))
     implementation testFixtures(project(':metadata'))
-    implementation project(':server').sourceSets.test.output
 
     implementation libs.jmhCore
     annotationProcessor libs.jmhGeneratorAnnProcess
@@ -4117,8 +4141,10 @@ gradle.projectsEvaluated {
   def protectedModules = [
     'clients': ':clients',
     'server-common': ':server-common',
+    'server': ':server',
     'storage/api': ':storage:storage-api',
     'coordinator-common': ':coordinator-common',
+    'storage': ':storage',
     'group-coordinator': ':group-coordinator',
     'share-coordinator': ':share-coordinator',
     'metadata': ':metadata',

checkstyle/import-control-server-common.xml

@@ -134,6 +134,7 @@
 
             <allow class="org.apache.kafka.server.util.TopicFilter.IncludeList" />
             <allow class="org.apache.kafka.test.TestUtils" />
+            <allow class="org.apache.kafka.server.authorizer.Authorizer" />
             <!-- ServerTestUtils uses yammer metrics for test cleanup -->
             <allow pkg="org.apache.kafka.server.metrics" />
             <allow pkg="com.yammer.metrics.core" />

checkstyle/import-control-server.xml

@@ -114,10 +114,12 @@
     <allow pkg="org.apache.kafka.network" />
     <allow pkg="org.apache.kafka.server" />
     <allow pkg="org.apache.kafka.server.authorizer" />
+    <allow class="org.apache.kafka.controller.MockAclMutator" />
   </subpackage>
 
   <subpackage name="network">
     <allow pkg="com.fasterxml.jackson" />
+    <allow pkg="org.apache.kafka.network.metrics" />
   </subpackage>
 
 </import-control>

checkstyle/import-control-storage.xml

@@ -118,6 +118,7 @@
         <allow pkg="org.apache.kafka.tiered.storage" />
 
         <allow pkg="kafka.api" />
+        <allow pkg="kafka.integration" />
         <allow pkg="kafka.log" />
         <allow pkg="kafka.server" />
         <allow pkg="kafka.utils" />

clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java

@@ -142,6 +142,34 @@ public synchronized void rebalance(Collection<TopicPartition> newAssignment) {
         this.subscriptions.rebalanceListener().ifPresent(crl -> crl.onPartitionsAssigned(added));
     }
 
+    /**
+     * Simulates a partition loss event. Calls {@link ConsumerRebalanceListener#onPartitionsLost}
+     * for the specified partitions and removes them from the current assignment. Unlike
+     * {@link #rebalance(Collection)}, which calls {@link ConsumerRebalanceListener#onPartitionsRevoked},
+     * this method models the case where the consumer loses partitions without a graceful revoke..
+     *
+     * <p>Only records belonging to the lost partitions are cleared; records for retained
+     * partitions are unaffected.
+     *
+     * @param partitionsLost the partitions to lose; all must be currently assigned
+     * @throws IllegalStateException if any partition is not currently assigned
+     */
+    public synchronized void losePartitions(Collection<TopicPartition> partitionsLost) {
+        Set<TopicPartition> currentAssignment = this.subscriptions.assignedPartitions();
+        Set<TopicPartition> lost = new HashSet<>(partitionsLost);
+        List<TopicPartition> notAssigned = lost.stream()
+            .filter(tp -> !currentAssignment.contains(tp))
+            .collect(Collectors.toList());
+        if (!notAssigned.isEmpty())
+            throw new IllegalStateException("Cannot lose partitions that are not currently assigned: " + notAssigned);
+        lost.forEach(records::remove);
+        this.subscriptions.rebalanceListener().ifPresent(crl -> crl.onPartitionsLost(lost));
+        Set<TopicPartition> remaining = currentAssignment.stream()
+            .filter(tp -> !lost.contains(tp))
+            .collect(Collectors.toSet());
+        this.subscriptions.assignFromSubscribed(remaining);
+    }
+
     @Override
     public synchronized Set<String> subscription() {
         return this.subscriptions.subscription();

clients/src/main/java/org/apache/kafka/common/security/ssl/DefaultSslEngineFactory.java

@@ -460,7 +460,7 @@ public boolean modified() {
 
         private KeyStore createKeyStoreFromPem(String privateKeyPem, String certChainPem, char[] keyPassword) {
             try {
-                KeyStore ks = KeyStore.getInstance("PKCS12");
+                KeyStore ks = KeyStore.getInstance(KeyStore.getDefaultType());
                 ks.load(null, null);
                 Key key = privateKey(privateKeyPem, keyPassword);
                 Certificate[] certChain = certs(certChainPem);
@@ -473,7 +473,7 @@ private KeyStore createKeyStoreFromPem(String privateKeyPem, String certChainPem
 
         private KeyStore createTrustStoreFromPem(String trustedCertsPem) {
             try {
-                KeyStore ts = KeyStore.getInstance("PKCS12");
+                KeyStore ts = KeyStore.getInstance(KeyStore.getDefaultType());
                 ts.load(null, null);
                 Certificate[] certs = certs(trustedCertsPem);
                 for (int i = 0; i < certs.length; i++) {

clients/src/test/java/org/apache/kafka/clients/consumer/MockConsumerTest.java

@@ -32,6 +32,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Optional;
+import java.util.Set;
 import java.util.stream.IntStream;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -230,4 +231,108 @@ public void shouldReturnMaxPollRecords() {
         assertTrue(records.isEmpty());
     }
 
+    @Test
+    public void testLosePartitionsCallsOnPartitionsLost() {
+        TopicPartition tp0 = new TopicPartition("test", 0);
+        TopicPartition tp1 = new TopicPartition("test", 1);
+        List<TopicPartition> assigned = List.of(tp0, tp1);
+
+        List<TopicPartition> lost = new ArrayList<>();
+        List<TopicPartition> revoked = new ArrayList<>();
+        consumer.subscribe(List.of("test"), new ConsumerRebalanceListener() {
+            @Override
+            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
+                revoked.addAll(partitions);
+            }
+            @Override
+            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {}
+            @Override
+            public void onPartitionsLost(Collection<TopicPartition> partitions) {
+                lost.addAll(partitions);
+            }
+        });
+
+        consumer.rebalance(assigned);
+        consumer.losePartitions(List.of(tp0));
+
+        assertEquals(List.of(tp0), lost);
+        assertTrue(revoked.isEmpty());
+    }
+
+    @Test
+    public void testLosePartitionsRemovesFromAssignment() {
+        TopicPartition tp0 = new TopicPartition("test", 0);
+        TopicPartition tp1 = new TopicPartition("test", 1);
+
+        consumer.subscribe(List.of("test"));
+        consumer.rebalance(List.of(tp0, tp1));
+        consumer.losePartitions(List.of(tp0));
+
+        assertFalse(consumer.assignment().contains(tp0));
+        assertTrue(consumer.assignment().contains(tp1));
+    }
+
+    @Test
+    public void testLosePartitionsThrowsIfNotAssigned() {
+        TopicPartition tp0 = new TopicPartition("test", 0);
+        TopicPartition tp1 = new TopicPartition("test", 1);
+
+        consumer.subscribe(List.of("test"));
+        consumer.rebalance(List.of(tp0));
+
+        assertThrows(IllegalStateException.class,
+                () -> consumer.losePartitions(List.of(tp1)));
+    }
+
+    @Test
+    public void testLosePartitionsClearsOnlyLostRecords() {
+        TopicPartition tp0 = new TopicPartition("test", 0);
+        TopicPartition tp1 = new TopicPartition("test", 1);
+
+        consumer.subscribe(List.of("test"));
+        consumer.rebalance(List.of(tp0, tp1));
+        consumer.updateBeginningOffsets(new HashMap<>() {{
+                put(tp0, 0L);
+                put(tp1, 0L);
+            }});
+        consumer.seek(tp0, 0);
+        consumer.seek(tp1, 0);
+
+        consumer.addRecord(new ConsumerRecord<>("test", 0, 0, null, null));
+        consumer.addRecord(new ConsumerRecord<>("test", 1, 0, null, null));
+
+        consumer.losePartitions(List.of(tp0));
+
+        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1));
+        assertEquals(1, records.count());
+
+        var record = records.iterator().next();
+        assertEquals(tp1, new TopicPartition(record.topic(), record.partition()));
+    }
+
+    @Test
+    public void testLosePartitionsThenRebalance() {
+        TopicPartition tp0 = new TopicPartition("test", 0);
+        TopicPartition tp1 = new TopicPartition("test", 1);
+        TopicPartition tp2 = new TopicPartition("test", 2);
+
+        List<TopicPartition> assigned = new ArrayList<>();
+        consumer.subscribe(List.of("test"), new ConsumerRebalanceListener() {
+            @Override
+            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {}
+            @Override
+            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+                assigned.addAll(partitions);
+            }
+        });
+
+        consumer.rebalance(List.of(tp0, tp1));
+        assigned.clear();
+
+        consumer.losePartitions(List.of(tp0));
+        consumer.rebalance(Arrays.asList(tp1, tp2));
+
+        assertEquals(List.of(tp2), assigned);
+        assertEquals(Set.of(tp1, tp2), consumer.assignment());
+    }
 }

clients/src/test/java/org/apache/kafka/common/security/ssl/DefaultSslEngineFactoryTest.java

@@ -221,6 +221,7 @@ public void testPemTrustStoreConfigWithOneCert() throws Exception {
         assertEquals(List.of("kafka0"), aliases);
         assertNotNull(trustStore.getCertificate("kafka0"), "Certificate not loaded");
         assertNull(trustStore.getKey("kafka0", null), "Unexpected private key");
+        assertEquals(KeyStore.getDefaultType(), trustStore.getType());
     }
 
     @Test
@@ -236,6 +237,7 @@ public void testPemTrustStoreConfigWithMultipleCerts() throws Exception {
         assertNull(trustStore.getKey("kafka0", null), "Unexpected private key");
         assertNotNull(trustStore.getCertificate("kafka1"), "Certificate not loaded");
         assertNull(trustStore.getKey("kafka1", null), "Unexpected private key");
+        assertEquals(KeyStore.getDefaultType(), trustStore.getType());
     }
 
     @Test
@@ -276,6 +278,7 @@ private void verifyPemKeyStoreConfig(String keyFileName, Password keyPassword) t
         assertNotNull(keyStore.getCertificate("kafka"), "Certificate not loaded");
         assertNotNull(keyStore.getKey("kafka", keyPassword == null ? null : keyPassword.value().toCharArray()),
             "Private key not loaded");
+        assertEquals(KeyStore.getDefaultType(), keyStore.getType());
     }
 
     @Test
@@ -289,6 +292,7 @@ public void testPemTrustStoreFile() throws Exception {
         assertEquals(List.of("kafka0"), aliases);
         assertNotNull(trustStore.getCertificate("kafka0"), "Certificate not found");
         assertNull(trustStore.getKey("kafka0", null), "Unexpected private key");
+        assertEquals(KeyStore.getDefaultType(), trustStore.getType());
     }
 
     @Test
@@ -304,6 +308,7 @@ public void testPemKeyStoreFileNoKeyPassword() throws Exception {
         assertEquals(List.of("kafka"), aliases);
         assertNotNull(keyStore.getCertificate("kafka"), "Certificate not loaded");
         assertNotNull(keyStore.getKey("kafka", null), "Private key not loaded");
+        assertEquals(KeyStore.getDefaultType(), keyStore.getType());
     }
 
     @Test
@@ -319,6 +324,7 @@ public void testPemKeyStoreFileWithKeyPassword() throws Exception {
         assertEquals(List.of("kafka"), aliases);
         assertNotNull(keyStore.getCertificate("kafka"), "Certificate not found");
         assertNotNull(keyStore.getKey("kafka", KEY_PASSWORD.value().toCharArray()), "Private key not found");
+        assertEquals(KeyStore.getDefaultType(), keyStore.getType());
     }
 
     private String pemFilePath(String pem) throws Exception {