Merge branch 'apache:trunk' into trunk

Author: Phuc-Hong-Tran

Diff for: build.gradle

@@ -208,7 +208,6 @@ if (repo != null) {
         'licenses/*',
         '**/generated/**',
         'clients/src/test/resources/serializedData/*',
-        'docker/resources/utility/go.sum',
         'docker/test/fixtures/secrets/*',
         'docker/examples/fixtures/secrets/*'
     ])
@@ -303,7 +302,7 @@ subprojects {
             from components.java
           } else {
             apply plugin: 'com.github.johnrengelman.shadow'
-            project.shadow.component(mavenJava)
+            artifact shadowJar
           }
 
           afterEvaluate {
@@ -1582,6 +1581,7 @@ project(':raft') {
     implementation libs.jacksonDatabind
 
     testImplementation project(':server-common')
+    testImplementation project(':server-common').sourceSets.test.output
     testImplementation project(':clients')
     testImplementation project(':clients').sourceSets.test.output
     testImplementation libs.junitJupiter

Diff for: checkstyle/import-control.xml

@@ -441,6 +441,7 @@
     <allow pkg="org.apache.kafka.common.protocol" />
     <allow pkg="org.apache.kafka.server.common" />
     <allow pkg="org.apache.kafka.server.common.serialization" />
+    <allow pkg="org.apache.kafka.server.fault"/>
     <allow pkg="org.apache.kafka.server.util" />
     <allow pkg="org.apache.kafka.test"/>
     <allow pkg="com.fasterxml.jackson" />

Diff for: clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java

@@ -1017,7 +1017,13 @@ synchronized public DescribeLogDirsResult describeLogDirs(Collection<Integer> br
                 for (Node node : nodes) {
                     Map<String, LogDirDescription> logDirDescriptionMap = unwrappedResults.get(node.id());
                     LogDirDescription logDirDescription = logDirDescriptionMap.getOrDefault(partitionLogDirs.get(0), new LogDirDescription(null, new HashMap<>()));
-                    logDirDescription.replicaInfos().put(new TopicPartition(topicName, topicPartitionInfo.partition()), new ReplicaInfo(0, 0, false));
+                    Map<TopicPartition, ReplicaInfo> topicPartitionReplicaInfoMap = new HashMap<>(logDirDescription.replicaInfos());
+                    topicPartitionReplicaInfoMap.put(new TopicPartition(topicName, topicPartitionInfo.partition()), new ReplicaInfo(0, 0, false));
+                    logDirDescriptionMap.put(partitionLogDirs.get(0), new LogDirDescription(
+                        logDirDescription.error(),
+                        topicPartitionReplicaInfoMap,
+                        logDirDescription.totalBytes().orElse(DescribeLogDirsResponse.UNKNOWN_VOLUME_BYTES),
+                        logDirDescription.usableBytes().orElse(DescribeLogDirsResponse.UNKNOWN_VOLUME_BYTES)));
                 }
             }
         }

Diff for: connect/basic-auth-extension/src/main/java/org/apache/kafka/connect/rest/basic/auth/extension/BasicAuthSecurityRestExtension.java

@@ -25,7 +25,6 @@
 import org.slf4j.LoggerFactory;
 
 import javax.security.auth.login.Configuration;
-import java.io.IOException;
 import java.util.Map;
 import java.util.function.Supplier;
 
@@ -100,7 +99,7 @@ public void register(ConnectRestExtensionContext restPluginContext) {
     }
 
     @Override
-    public void close() throws IOException {
+    public void close() {
 
     }
 

Diff for: connect/basic-auth-extension/src/main/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilter.java

@@ -34,15 +34,13 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.Base64;
 
 import javax.security.auth.callback.Callback;
 import javax.security.auth.callback.CallbackHandler;
 import javax.security.auth.callback.NameCallback;
 import javax.security.auth.callback.PasswordCallback;
-import javax.security.auth.callback.UnsupportedCallbackException;
 import javax.security.auth.login.LoginContext;
 import javax.security.auth.login.LoginException;
 import javax.ws.rs.Priorities;
@@ -87,7 +85,7 @@ public JaasBasicAuthFilter(Configuration configuration) {
     }
 
     @Override
-    public void filter(ContainerRequestContext requestContext) throws IOException {
+    public void filter(ContainerRequestContext requestContext) {
         if (isInternalRequest(requestContext)) {
             log.trace("Skipping authentication for internal request");
             return;
@@ -119,16 +117,16 @@ private boolean isInternalRequest(ContainerRequestContext requestContext) {
 
     public static class BasicAuthCallBackHandler implements CallbackHandler {
 
-        private String username;
-        private String password;
+        private final String username;
+        private final String password;
 
         public BasicAuthCallBackHandler(BasicAuthCredentials credentials) {
             username = credentials.username();
             password = credentials.password();
         }
 
         @Override
-        public void handle(Callback[] callbacks) throws UnsupportedCallbackException {
+        public void handle(Callback[] callbacks) {
             List<Callback> unsupportedCallbacks = new ArrayList<>();
             for (Callback callback : callbacks) {
                 if (callback instanceof NameCallback) {

Diff for: connect/basic-auth-extension/src/main/java/org/apache/kafka/connect/rest/basic/auth/extension/PropertyFileLoginModule.java

@@ -54,7 +54,7 @@ public class PropertyFileLoginModule implements LoginModule {
     private String fileName;
     private boolean authenticated;
 
-    private static Map<String, Properties> credentialPropertiesMap = new ConcurrentHashMap<>();
+    private static final Map<String, Properties> CREDENTIAL_PROPERTIES = new ConcurrentHashMap<>();
 
     @Override
     public void initialize(Subject subject, CallbackHandler callbackHandler, Map<String, ?> sharedState, Map<String, ?> options) {
@@ -64,15 +64,15 @@ public void initialize(Subject subject, CallbackHandler callbackHandler, Map<Str
             throw new ConfigException("Property Credentials file must be specified");
         }
 
-        if (!credentialPropertiesMap.containsKey(fileName)) {
+        if (!CREDENTIAL_PROPERTIES.containsKey(fileName)) {
             log.trace("Opening credential properties file '{}'", fileName);
             Properties credentialProperties = new Properties();
             try {
                 try (InputStream inputStream = Files.newInputStream(Paths.get(fileName))) {
                     log.trace("Parsing credential properties file '{}'", fileName);
                     credentialProperties.load(inputStream);
                 }
-                credentialPropertiesMap.putIfAbsent(fileName, credentialProperties);
+                CREDENTIAL_PROPERTIES.putIfAbsent(fileName, credentialProperties);
                 if (credentialProperties.isEmpty())
                     log.warn("Credential properties file '{}' is empty; all requests will be permitted",
                         fileName);
@@ -101,7 +101,7 @@ public boolean login() throws LoginException {
         String username = ((NameCallback) callbacks[0]).getName();
         char[] passwordChars = ((PasswordCallback) callbacks[1]).getPassword();
         String password = passwordChars != null ? new String(passwordChars) : null;
-        Properties credentialProperties = credentialPropertiesMap.get(fileName);
+        Properties credentialProperties = CREDENTIAL_PROPERTIES.get(fileName);
 
         if (credentialProperties.isEmpty()) {
             log.trace("Not validating credentials for user '{}' as credential properties file '{}' is empty",
@@ -132,17 +132,17 @@ public boolean login() throws LoginException {
     }
 
     @Override
-    public boolean commit() throws LoginException {
+    public boolean commit() {
         return authenticated;
     }
 
     @Override
-    public boolean abort() throws LoginException {
+    public boolean abort() {
         return true;
     }
 
     @Override
-    public boolean logout() throws LoginException {
+    public boolean logout() {
         return true;
     }
 

Diff for: connect/basic-auth-extension/src/test/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilterTest.java

@@ -131,7 +131,7 @@ public void testUnknownLoginModule() throws IOException {
     }
 
     @Test
-    public void testUnknownCredentialsFile() throws IOException {
+    public void testUnknownCredentialsFile() {
         JaasBasicAuthFilter jaasBasicAuthFilter = setupJaasFilter("KafkaConnect", "/tmp/testcrednetial");
         ContainerRequestContext requestContext = setMock("Basic", "user", "password");
         jaasBasicAuthFilter.filter(requestContext);
@@ -142,7 +142,7 @@ public void testUnknownCredentialsFile() throws IOException {
     }
 
     @Test
-    public void testNoFileOption() throws IOException {
+    public void testNoFileOption() {
         JaasBasicAuthFilter jaasBasicAuthFilter = setupJaasFilter("KafkaConnect", null);
         ContainerRequestContext requestContext = setMock("Basic", "user", "password");
         jaasBasicAuthFilter.filter(requestContext);

Diff for: connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSinkTaskTest.java

@@ -40,7 +40,6 @@ public class FileStreamSinkTaskTest {
 
     private FileStreamSinkTask task;
     private ByteArrayOutputStream os;
-    private PrintStream printStream;
 
     @TempDir
     public Path topDir;
@@ -49,7 +48,7 @@ public class FileStreamSinkTaskTest {
     @BeforeEach
     public void setup() {
         os = new ByteArrayOutputStream();
-        printStream = new PrintStream(os);
+        PrintStream printStream = new PrintStream(os);
         task = new FileStreamSinkTask(printStream);
         outputFile = topDir.resolve("connect.output").toAbsolutePath().toString();
     }

Diff for: core/src/main/scala/kafka/admin/ConfigCommand.scala

@@ -22,7 +22,7 @@ import java.util.concurrent.TimeUnit
 import java.util.{Collections, Properties}
 import joptsimple._
 import kafka.server.DynamicConfig.QuotaConfigs
-import kafka.server.{ConfigEntityName, Defaults, DynamicBrokerConfig, DynamicConfig, KafkaConfig}
+import kafka.server.{Defaults, DynamicBrokerConfig, DynamicConfig, KafkaConfig}
 import kafka.utils.{Exit, Logging, PasswordEncoder}
 import kafka.utils.Implicits._
 import kafka.zk.{AdminZkClient, KafkaZkClient}
@@ -35,7 +35,7 @@ import org.apache.kafka.common.quota.{ClientQuotaAlteration, ClientQuotaEntity,
 import org.apache.kafka.common.security.JaasUtils
 import org.apache.kafka.common.security.scram.internals.{ScramCredentialUtils, ScramFormatter, ScramMechanism}
 import org.apache.kafka.common.utils.{Sanitizer, Time, Utils}
-import org.apache.kafka.server.config.ConfigType
+import org.apache.kafka.server.config.{ConfigEntityName, ConfigType}
 import org.apache.kafka.server.util.{CommandDefaultOptions, CommandLineUtils}
 import org.apache.kafka.storage.internals.log.LogConfig
 import org.apache.zookeeper.client.ZKClientConfig
@@ -152,7 +152,7 @@ object ConfigCommand extends Logging {
       if (!configsToBeAdded.isEmpty || configsToBeDeleted.nonEmpty) {
         validateBrokersNotRunning(entityName, adminZkClient, zkClient, errorMessage)
 
-        val perBrokerConfig = entityName != ConfigEntityName.Default
+        val perBrokerConfig = entityName != ConfigEntityName.DEFAULT
         preProcessBrokerConfigs(configsToBeAdded, perBrokerConfig)
       }
     }
@@ -177,7 +177,7 @@ object ConfigCommand extends Logging {
                                         adminZkClient: AdminZkClient,
                                         zkClient: KafkaZkClient,
                                         errorMessage: String): Unit = {
-    val perBrokerConfig = entityName != ConfigEntityName.Default
+    val perBrokerConfig = entityName != ConfigEntityName.DEFAULT
     val info = "Broker configuration operations using ZooKeeper are only supported if the affected broker(s) are not running."
     if (perBrokerConfig) {
       adminZkClient.parseBroker(entityName).foreach { brokerId =>
@@ -696,7 +696,7 @@ object ConfigCommand extends Logging {
         case t => t
       }
       sanitizedName match {
-        case Some(ConfigEntityName.Default) => "default " + typeName
+        case Some(ConfigEntityName.DEFAULT) => "default " + typeName
         case Some(n) =>
           val desanitized = if (entityType == ConfigType.USER || entityType == ConfigType.CLIENT) Sanitizer.desanitize(n) else n
           s"$typeName '$desanitized'"
@@ -757,7 +757,7 @@ object ConfigCommand extends Logging {
     else {
       // Exactly one entity type and at-most one entity name expected for other entities
       val name = entityNames.headOption match {
-        case Some("") => Some(ConfigEntityName.Default)
+        case Some("") => Some(ConfigEntityName.DEFAULT)
         case v => v
       }
       ConfigEntity(Entity(entityTypes.head, name), None)
@@ -774,7 +774,7 @@ object ConfigCommand extends Logging {
 
     def sanitizeName(entityType: String, name: String) = {
       if (name.isEmpty)
-        ConfigEntityName.Default
+        ConfigEntityName.DEFAULT
       else {
         entityType match {
           case ConfigType.USER | ConfigType.CLIENT => Sanitizer.sanitize(name)