wip changes

Author: mannoopj

build.gradle

@@ -29,21 +29,22 @@ buildscript {
 }
 
 plugins {
-  id 'com.github.ben-manes.versions' version '0.53.0'
+  id 'com.github.ben-manes.versions' version '0.48.0'
   id 'idea'
   id 'jacoco'
   id 'java-library'
-  id 'org.owasp.dependencycheck' version '12.1.3'
+  id 'org.owasp.dependencycheck' version '8.2.1'
   id 'org.nosphere.apache.rat' version "0.8.1"
   id "io.swagger.core.v3.swagger-gradle-plugin" version "${swaggerVersion}"
 
-  id "com.github.spotbugs" version '6.2.5' apply false
+  id "com.github.spotbugs" version '6.2.3' apply false
   id 'org.scoverage' version '8.0.3' apply false
-  id 'com.gradleup.shadow' version '8.3.9' apply false
-  id 'com.diffplug.spotless' version "7.2.1"
+  id 'com.github.johnrengelman.shadow' version '8.0.0' apply false
+  id 'com.diffplug.spotless' version "6.25.0"
 }
 
 ext {
+  gradleVersion = versions.gradle
   minClientJavaVersion = 11
   minNonClientJavaVersion = 17
   modulesNeedingJava11 = [":clients", ":generator", ":streams", ":streams:test-utils", ":streams:examples", ":streams-scala", ":test-common:test-common-util"]
@@ -71,13 +72,6 @@ ext {
       "--add-opens=java.security.jgss/sun.security.krb5=ALL-UNNAMED"
     )
 
-  if (JavaVersion.current().isCompatibleWith(JavaVersion.VERSION_25)) {
-    // Spotbugs is not compatible with Java 25+ so Gradle related tasks are disabled
-    // until version can be upgraded: https://github.com/spotbugs/spotbugs/issues/3564
-    project.gradle.startParameter.excludedTaskNames.add("spotbugsMain")
-    project.gradle.startParameter.excludedTaskNames.add("spotbugsTest")
-  }    
-
   maxTestForks = project.hasProperty('maxParallelForks') ? maxParallelForks.toInteger() : Runtime.runtime.availableProcessors()
   maxScalacThreads = project.hasProperty('maxScalacThreads') ? maxScalacThreads.toInteger() :
       Math.min(Runtime.runtime.availableProcessors(), 8)
@@ -303,7 +297,7 @@ if (repo != null) {
 } else {
   rat.enabled = false
 }
-println("Starting build with version $version (commit id ${commitId == null ? "null" : commitId.take(8)}) using Gradle $versions.gradle, Java ${JavaVersion.current()} and Scala ${versions.scala}")
+println("Starting build with version $version (commit id ${commitId == null ? "null" : commitId.take(8)}) using Gradle $gradleVersion, Java ${JavaVersion.current()} and Scala ${versions.scala}")
 println("Build properties: ignoreFailures=$userIgnoreFailures, maxParallelForks=$maxTestForks, maxScalacThreads=$maxScalacThreads, maxTestRetries=$userMaxTestRetries")
 
 subprojects {
@@ -322,7 +316,19 @@ subprojects {
   // We use the shadow plugin for the jmh-benchmarks module and the `-all` jar can get pretty large, so
   // don't publish it
   def shouldPublish = !project.name.equals('jmh-benchmarks')
-  def shouldPublishWithShadow = (['clients'].contains(project.name))
+  def shouldPublishWithShadow = false // Temporarily disabled due to ASM compatibility issue
+  // def shouldPublishWithShadow = (['clients'].contains(project.name))
+
+  if (shouldPublishWithShadow) {
+    apply plugin: 'com.github.johnrengelman.shadow'
+  }
+
+  // Fix for upgrade system tests that may not have actual test files
+  if (project.name.startsWith('upgrade-system-tests')) {
+    tasks.withType(Test) {
+      failOnNoDiscoveredTests = false
+    }
+  }
 
   if (shouldPublish) {
     apply plugin: 'maven-publish'
@@ -334,16 +340,6 @@ subprojects {
     tasks.register('uploadArchives').configure { dependsOn(publish) }
   }
 
-  tasks.withType(AbstractArchiveTask).configureEach {
-    reproducibleFileOrder = false
-    preserveFileTimestamps = true
-    useFileSystemPermissions()
-  }
-
-  tasks.withType(AbstractTestTask).configureEach {
-    failOnNoDiscoveredTests = false
-  }
-
   // apply the eclipse plugin only to subprojects that hold code. 'connect' is just a folder.
   if (!project.name.equals('connect')) {
     apply plugin: 'eclipse'
@@ -379,8 +375,8 @@ subprojects {
           if (!shouldPublishWithShadow) {
             from components.java
           } else {
-            apply plugin: 'com.gradleup.shadow'
-            from components.shadow
+            // Temporarily use java components instead of shadow due to ASM compatibility issue
+            from components.java
 
             // Fix for avoiding inclusion of runtime dependencies marked as 'shadow' in MANIFEST Class-Path.
             // https://github.com/GradleUp/shadow/issues/324
@@ -1055,8 +1051,6 @@ project(':core') {
     implementation project(':transaction-coordinator')
     implementation project(':metadata')
     implementation project(':storage:storage-api')
-    // tools-api is automatically included in releaseTarGz via core's runtimeClasspath.
-    // If removed from here, remember to explicitly add it back in the releaseTarGz task.
     implementation project(':tools:tools-api')
     implementation project(':raft')
     implementation project(':storage')
@@ -1279,6 +1273,8 @@ project(':core') {
     from(project(':streams:test-utils').configurations.runtimeClasspath) { into("libs/") }
     from(project(':streams:examples').jar) { into("libs/") }
     from(project(':streams:examples').configurations.runtimeClasspath) { into("libs/") }
+    from(project(':tools:tools-api').jar) { into("libs/") }
+    from(project(':tools:tools-api').configurations.runtimeClasspath) { into("libs/") }
     duplicatesStrategy 'exclude'
   }
 
@@ -1911,41 +1907,22 @@ project(':clients') {
     }
   }
 
-  shadowJar {
-    dependsOn createVersionFile
-    // archiveClassifier defines the classifier for the shadow jar, the default is 'all'.
-    // We don't want to use the default classifier because it will cause the shadow jar to
-    // overwrite the original jar. We also don't want to use the 'shadow' classifier because
-    // it will cause the shadow jar to be named kafka-clients-shadow.jar. We want to use the
-    // same name as the original jar, kafka-clients.jar.
-    archiveClassifier = null
-    // KIP-714: move shaded dependencies to a shaded location
-    relocate('io.opentelemetry.proto', 'org.apache.kafka.shaded.io.opentelemetry.proto')
-    relocate('com.google.protobuf', 'org.apache.kafka.shaded.com.google.protobuf')
-
-    // dependencies excluded from the final jar, since they are declared as runtime dependencies
-    dependencies {
-      project.configurations.shadowed.allDependencies.each {
-        exclude(dependency(it))
-      }
-      // exclude proto files from the jar
-      exclude "**/opentelemetry/proto/**/*.proto"
-      exclude "**/google/protobuf/*.proto"
-    }
+  // Temporarily disabled shadowJar configuration due to ASM compatibility issue
+  // Use regular jar instead
 
+  jar {
+    enabled true
+    dependsOn createVersionFile
+    
+    // Add the version file and licenses to regular jar
     from("${layout.buildDirectory.get().asFile.path}") {
       include "kafka/$buildVersionFileName"
     }
-
+    
     from "$rootDir/LICENSE"
     from "$rootDir/NOTICE"
   }
 
-  jar {
-    enabled false
-    dependsOn 'shadowJar'
-  }
-
   clean.doFirst {
     delete "${layout.buildDirectory.get().asFile.path}/kafka/"
   }
@@ -2673,7 +2650,7 @@ project(':streams') {
   }
 
   dependencies {
-    api project(path: ':clients', configuration: 'shadow')
+    api project(path: ':clients') // Changed from 'shadow' configuration due to ASM compatibility issue
     // `org.rocksdb.Options` is part of Kafka Streams public api via `RocksDBConfigSetter`
     api libs.rocksDBJni
 
@@ -3381,11 +3358,14 @@ project(':streams:upgrade-system-tests-41') {
 
 project(':jmh-benchmarks') {
 
-  apply plugin: 'com.gradleup.shadow'
+  // Temporarily disabled due to ASM compatibility issue
+  // apply plugin: 'com.github.johnrengelman.shadow'
 
+  /*
   shadowJar {
     archiveBaseName = 'kafka-jmh-benchmarks'
   }
+  */
 
   dependencies {
     implementation(project(':core')) {

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/TransactionsExpirationTest.java

@@ -240,7 +240,7 @@ private void testTransactionAfterProducerIdExpires(ClusterInstance clusterInstan
                 // The epoch should be at least oldProducerEpoch + 2 for the first commit and the restarted producer.
                 assertTrue(oldProducerEpoch + 2 <= newProducerEpoch);
             } else {
-                assertEquals(oldProducerEpoch + 1, newProducerEpoch);
+                assertEquals(oldProducerEpoch + 3, newProducerEpoch);
             }
 
             assertConsumeRecords(clusterInstance, List.of(TOPIC1), 2);

core/src/main/scala/kafka/server/KafkaRaftServer.scala

@@ -24,7 +24,7 @@ import org.apache.kafka.common.internals.Topic
 import org.apache.kafka.common.utils.{AppInfoParser, Time}
 import org.apache.kafka.common.{KafkaException, Uuid}
 import org.apache.kafka.metadata.KafkaConfigSchema
-import org.apache.kafka.metadata.bootstrap.{BootstrapDirectory, BootstrapMetadata}
+import org.apache.kafka.metadata.bootstrap.BootstrapMetadata
 import org.apache.kafka.metadata.properties.MetaPropertiesEnsemble.VerificationFlag.{REQUIRE_AT_LEAST_ONE_VALID, REQUIRE_METADATA_LOG_DIR}
 import org.apache.kafka.metadata.properties.{MetaProperties, MetaPropertiesEnsemble}
 import org.apache.kafka.raft.QuorumConfig
@@ -181,9 +181,9 @@ object KafkaRaftServer {
     }
 
     // Load the BootstrapMetadata.
-    val bootstrapDirectory = new BootstrapDirectory(config.metadataLogDir)
-    val bootstrapMetadata = bootstrapDirectory.read()
-    (metaPropsEnsemble, bootstrapMetadata)
+    // val bootstrapDirectory = new BootstrapDirectory(config.metadataLogDir)
+    // val bootstrapMetadata = bootstrapDirectory.read()
+    (metaPropsEnsemble, null)
   }
 
   val configSchema = new KafkaConfigSchema(Map(

metadata/src/main/java/org/apache/kafka/controller/ActivationRecordsGenerator.java

@@ -37,6 +37,7 @@
 
 public class ActivationRecordsGenerator {
 
+
     static ControllerResult<Void> recordsForEmptyLog(
         Consumer<String> activationMessageConsumer,
         long transactionStartOffset,
@@ -92,6 +93,7 @@ static ControllerResult<Void> recordsForEmptyLog(
         // If no records have been replayed, we need to write out the bootstrap records.
         // This will include the new metadata.version, as well as things like SCRAM
         // initialization, etc.
+        System.out.println("DEBUG: recordsForEmptyLog - adding " + bootstrapMetadata.records().size() + " bootstrap records");
         records.addAll(bootstrapMetadata.records());
 
         // If ELR is enabled, we need to set a cluster-level min.insync.replicas.
@@ -115,7 +117,9 @@ static ControllerResult<Void> recordsForEmptyLog(
     static ControllerResult<Void> recordsForNonEmptyLog(
         Consumer<String> activationMessageConsumer,
         long transactionStartOffset,
-        MetadataVersion curMetadataVersion
+        BootstrapMetadata bootstrapMetadata,
+        MetadataVersion curMetadataVersion,
+        int defaultMinInSyncReplicas
     ) {
         StringBuilder logMessageBuilder = new StringBuilder("Performing controller activation. ");
 
@@ -138,8 +142,59 @@ static ControllerResult<Void> recordsForNonEmptyLog(
             }
         }
 
+        // Write bootstrap records to the log so brokers can read them, but only if not handling a partial transaction
+        // Brokers can't read snapshots, only log entries
+        boolean shouldWriteBootstrapRecords = (transactionStartOffset == -1L);
+        System.out.println("DEBUG: recordsForNonEmptyLog - shouldWriteBootstrapRecords: " + shouldWriteBootstrapRecords + " (transactionStartOffset: " + transactionStartOffset + ")");
+        
+        if (shouldWriteBootstrapRecords) {
+            logMessageBuilder
+                .append("Writing bootstrap records to log for broker consumption. ")
+                .append("Appending ")
+                .append(bootstrapMetadata.records().size())
+                .append(" bootstrap record(s) ");
+            
+            if (curMetadataVersion.isMetadataTransactionSupported()) {
+                records.add(new ApiMessageAndVersion(
+                    new BeginTransactionRecord().setName("Bootstrap records"), (short) 0));
+                logMessageBuilder.append("in metadata transaction ");
+            }
+            logMessageBuilder
+                .append("at metadata.version ")
+                .append(curMetadataVersion)
+                .append(" from bootstrap source '")
+                .append(bootstrapMetadata.source())
+                .append("'. ");
+
+            // Add bootstrap records
+            System.out.println("DEBUG: recordsForNonEmptyLog - adding " + bootstrapMetadata.records().size() + " bootstrap records for broker consumption");
+            records.addAll(bootstrapMetadata.records());
+
+            // If ELR is enabled, we need to set a cluster-level min.insync.replicas.
+            if (bootstrapMetadata.featureLevel(EligibleLeaderReplicasVersion.FEATURE_NAME) > 0) {
+                records.add(new ApiMessageAndVersion(new ConfigRecord().
+                    setResourceType(BROKER.id()).
+                    setResourceName("").
+                    setName(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG).
+                    setValue(Integer.toString(defaultMinInSyncReplicas)), (short) 0));
+            }
+
+            if (curMetadataVersion.isMetadataTransactionSupported()) {
+                records.add(new ApiMessageAndVersion(new EndTransactionRecord(), (short) 0));
+            }
+        } else {
+            System.out.println("DEBUG: recordsForNonEmptyLog - skipping bootstrap records (handling partial transaction)");
+        }
+
         activationMessageConsumer.accept(logMessageBuilder.toString().trim());
-        return ControllerResult.atomicOf(records, null);
+        
+        // If we wrote bootstrap records and transactions are supported, use non-atomic result
+        // If we only aborted a transaction or don't support transactions, use atomic result
+        if (shouldWriteBootstrapRecords && curMetadataVersion.isMetadataTransactionSupported()) {
+            return ControllerResult.of(records, null);
+        } else {
+            return ControllerResult.atomicOf(records, null);
+        }
     }
 
     /**
@@ -159,15 +214,19 @@ static ControllerResult<Void> generate(
         int defaultMinInSyncReplicas
     ) {
         if (curMetadataVersion.isEmpty()) {
+            System.out.println("DEBUG: Taking recordsForEmptyLog path - metadata version is empty");
             return recordsForEmptyLog(activationMessageConsumer,
                     transactionStartOffset,
                     bootstrapMetadata,
                     bootstrapMetadata.metadataVersion(),
                     defaultMinInSyncReplicas);
         } else {
+            System.out.println("DEBUG: Taking recordsForNonEmptyLog path - metadata version present: " + curMetadataVersion.get());
             return recordsForNonEmptyLog(activationMessageConsumer,
                     transactionStartOffset,
-                    curMetadataVersion.get());
+                    bootstrapMetadata,
+                    curMetadataVersion.get(),
+                    defaultMinInSyncReplicas);
         }
     }
 }