From f63cebef4456b69e056d46a54a3c808438f6234f Mon Sep 17 00:00:00 2001
From: Luke Chen <showuon@gmail.com>
Date: Sun, 24 Oct 2021 11:05:13 +0800
Subject: [PATCH] KAFKA-13396: allow create topic without
 partition/replicaFactor

---
 .../main/scala/kafka/admin/TopicCommand.scala |  2 -
 .../unit/kafka/admin/TopicCommandTest.scala   | 47 +++++++++++++------
 2 files changed, 33 insertions(+), 16 deletions(-)

diff --git a/core/src/main/scala/kafka/admin/TopicCommand.scala b/core/src/main/scala/kafka/admin/TopicCommand.scala
index f7b1c870b9cdd..5e7d98c1d2ed5 100755
--- a/core/src/main/scala/kafka/admin/TopicCommand.scala
+++ b/core/src/main/scala/kafka/admin/TopicCommand.scala
@@ -625,8 +625,6 @@ object TopicCommand extends Logging {
       }
       if (!has(listOpt) && !has(describeOpt))
         CommandLineUtils.checkRequiredArgs(parser, options, topicOpt)
-      if (has(createOpt) && !has(replicaAssignmentOpt))
-        CommandLineUtils.checkRequiredArgs(parser, options, partitionsOpt, replicationFactorOpt)
       if (has(alterOpt)) {
         CommandLineUtils.checkInvalidArgsSet(parser, options, Set(bootstrapServerOpt, configOpt), Set(alterOpt),
         Some(kafkaConfigsCanAlterTopicConfigsViaBootstrapServer))
diff --git a/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala b/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala
index f34e154890509..9586cf5395c46 100644
--- a/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/TopicCommandTest.scala
@@ -82,23 +82,23 @@ class TopicCommandTest {
   }
 
   @Test
-  def testCreateWithPartitionCountWithoutReplicationFactor(): Unit = {
-    assertCheckArgsExitCode(1,
-      new TopicCommandOptions(
-        Array("--bootstrap-server", brokerList,
-          "--create",
-          "--partitions", "2",
-          "--topic", topicName)))
+  def testCreateWithPartitionCountWithoutReplicationFactorShouldSucceed(): Unit = {
+    val opts = new TopicCommandOptions(
+      Array("--bootstrap-server", brokerList,
+        "--create",
+        "--partitions", "2",
+        "--topic", topicName))
+    opts.checkArgs()
   }
 
   @Test
-  def testCreateWithReplicationFactorWithoutPartitionCount(): Unit = {
-    assertCheckArgsExitCode(1,
-      new TopicCommandOptions(
-        Array("--bootstrap-server", brokerList,
-          "--create",
-          "--replication-factor", "3",
-          "--topic", topicName)))
+  def testCreateWithReplicationFactorWithoutPartitionCountShouldSucceed(): Unit = {
+    val opts = new TopicCommandOptions(
+      Array("--bootstrap-server", brokerList,
+        "--create",
+        "--replication-factor", "3",
+        "--topic", topicName))
+    opts.checkArgs()
   }
 
   @Test
@@ -123,6 +123,25 @@ class TopicCommandTest {
           "--topic", topicName)))
   }
 
+  @Test
+  def testCreateWithoutPartitionCountAndReplicationFactorShouldSucceed(): Unit = {
+    val opts = new TopicCommandOptions(
+      Array("--bootstrap-server", brokerList,
+        "--create",
+        "--topic", topicName))
+    opts.checkArgs()
+  }
+
+  @Test
+  def testDescribeShouldSucceed(): Unit = {
+    val opts = new TopicCommandOptions(
+      Array("--bootstrap-server", brokerList,
+        "--describe",
+        "--topic", topicName))
+    opts.checkArgs()
+  }
+
+
   @Test
   def testParseAssignmentDuplicateEntries(): Unit = {
     assertThrows(classOf[AdminCommandFailedException], () => TopicCommand.parseReplicaAssignment("5:5"))