mirror of https://github.com/apache/kafka.git
KAFKA-14711: kafaka-metadata-quorum.sh does not honor --command-confi… (#13241)
…g option https://github.com/apache/kafka/pull/12951 accidentally changed the behavior of the `kafaka-metadata-quorum.sh` CLI by making it silently ignore a `--command-config <filename>` properties file that exists. This was an undetected regression in the 3.4.0 release. This patch fixes the issue such that any such specified file will be honored. Reviewers: José Armando García Sancio <jsancio@apache.org>, Ismael Juma <ismael@juma.me.uk>
This commit is contained in:
parent
f7a9f5d6a3
commit
b09edcdc9e
|
@ -398,6 +398,7 @@
|
||||||
<allow pkg="org.apache.kafka.clients.admin" />
|
<allow pkg="org.apache.kafka.clients.admin" />
|
||||||
<allow pkg="org.apache.kafka.clients.producer" />
|
<allow pkg="org.apache.kafka.clients.producer" />
|
||||||
<allow pkg="org.apache.kafka.clients.consumer" />
|
<allow pkg="org.apache.kafka.clients.consumer" />
|
||||||
|
<allow pkg="org.apache.kafka.test" />
|
||||||
<allow pkg="com.fasterxml.jackson" />
|
<allow pkg="com.fasterxml.jackson" />
|
||||||
<allow pkg="org.jose4j" />
|
<allow pkg="org.jose4j" />
|
||||||
<allow pkg="net.sourceforge.argparse4j" />
|
<allow pkg="net.sourceforge.argparse4j" />
|
||||||
|
|
|
@ -31,6 +31,7 @@ import org.apache.kafka.common.utils.Utils;
|
||||||
import org.apache.kafka.server.util.ToolsUtils;
|
import org.apache.kafka.server.util.ToolsUtils;
|
||||||
|
|
||||||
import java.io.File;
|
import java.io.File;
|
||||||
|
import java.io.IOException;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.Comparator;
|
import java.util.Comparator;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
|
@ -86,14 +87,8 @@ public class MetadataQuorumCommand {
|
||||||
Namespace namespace = parser.parseArgsOrFail(args);
|
Namespace namespace = parser.parseArgsOrFail(args);
|
||||||
String command = namespace.getString("command");
|
String command = namespace.getString("command");
|
||||||
|
|
||||||
File commandConfig = namespace.get("command_config");
|
File optionalCommandConfig = namespace.get("command_config");
|
||||||
Properties props = new Properties();
|
final Properties props = getProperties(optionalCommandConfig);
|
||||||
if (commandConfig != null) {
|
|
||||||
if (!commandConfig.exists())
|
|
||||||
throw new TerseException("Properties file " + commandConfig.getPath() + " does not exists!");
|
|
||||||
|
|
||||||
Utils.loadProps(commandConfig.getPath());
|
|
||||||
}
|
|
||||||
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, namespace.getString("bootstrap_server"));
|
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, namespace.getString("bootstrap_server"));
|
||||||
admin = Admin.create(props);
|
admin = Admin.create(props);
|
||||||
|
|
||||||
|
@ -116,6 +111,16 @@ public class MetadataQuorumCommand {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private static Properties getProperties(File optionalCommandConfig) throws TerseException, IOException {
|
||||||
|
if (optionalCommandConfig == null) {
|
||||||
|
return new Properties();
|
||||||
|
} else {
|
||||||
|
if (!optionalCommandConfig.exists())
|
||||||
|
throw new TerseException("Properties file " + optionalCommandConfig.getPath() + " does not exists!");
|
||||||
|
return Utils.loadProps(optionalCommandConfig.getPath());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private static void addDescribeParser(Subparsers subparsers) {
|
private static void addDescribeParser(Subparsers subparsers) {
|
||||||
Subparser describeParser = subparsers
|
Subparser describeParser = subparsers
|
||||||
.addParser("describe")
|
.addParser("describe")
|
||||||
|
|
|
@ -22,10 +22,14 @@ import kafka.test.annotation.ClusterTestDefaults;
|
||||||
import kafka.test.annotation.ClusterTests;
|
import kafka.test.annotation.ClusterTests;
|
||||||
import kafka.test.annotation.Type;
|
import kafka.test.annotation.Type;
|
||||||
import kafka.test.junit.ClusterTestExtensions;
|
import kafka.test.junit.ClusterTestExtensions;
|
||||||
|
import org.apache.kafka.clients.admin.AdminClientConfig;
|
||||||
import org.apache.kafka.common.errors.UnsupportedVersionException;
|
import org.apache.kafka.common.errors.UnsupportedVersionException;
|
||||||
|
import org.apache.kafka.test.TestUtils;
|
||||||
import org.junit.jupiter.api.Tag;
|
import org.junit.jupiter.api.Tag;
|
||||||
import org.junit.jupiter.api.extension.ExtendWith;
|
import org.junit.jupiter.api.extension.ExtendWith;
|
||||||
|
|
||||||
|
import java.io.File;
|
||||||
|
import java.io.IOException;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.concurrent.ExecutionException;
|
import java.util.concurrent.ExecutionException;
|
||||||
import java.util.regex.Pattern;
|
import java.util.regex.Pattern;
|
||||||
|
@ -141,6 +145,16 @@ class MetadataQuorumCommandTest {
|
||||||
assertEquals("0", replicationOutput.split("\n")[1].split("\\s+")[2]);
|
assertEquals("0", replicationOutput.split("\n")[1].split("\\s+")[2]);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ClusterTests({
|
||||||
|
@ClusterTest(clusterType = Type.CO_KRAFT, brokers = 1, controllers = 1)
|
||||||
|
})
|
||||||
|
public void testCommandConfig() throws IOException {
|
||||||
|
// specifying a --command-config containing properties that would prevent login must fail
|
||||||
|
File tmpfile = TestUtils.tempFile(AdminClientConfig.SECURITY_PROTOCOL_CONFIG + "=SSL_PLAINTEXT");
|
||||||
|
assertEquals(1, MetadataQuorumCommand.mainNoExit("--bootstrap-server", cluster.bootstrapServers(),
|
||||||
|
"--command-config", tmpfile.getAbsolutePath(), "describe", "--status"));
|
||||||
|
}
|
||||||
|
|
||||||
@ClusterTest(clusterType = Type.ZK, brokers = 1)
|
@ClusterTest(clusterType = Type.ZK, brokers = 1)
|
||||||
public void testDescribeQuorumInZkMode() {
|
public void testDescribeQuorumInZkMode() {
|
||||||
assertTrue(
|
assertTrue(
|
||||||
|
|
Loading…
Reference in New Issue