KAFKA-14577: Move ConsoleProducer to tools module

bin/kafka-console-producer.sh

@@ -17,4 +17,4 @@
 if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
     export KAFKA_HEAP_OPTS="-Xmx512M"
 fi
-exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleProducer "$@"
+exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.tools.ConsoleProducer "$@"

bin/windows/kafka-console-producer.bat

@@ -16,5 +16,5 @@ rem limitations under the License.
 
 SetLocal
 set KAFKA_HEAP_OPTS=-Xmx512M
-"%~dp0kafka-run-class.bat" kafka.tools.ConsoleProducer %*
+"%~dp0kafka-run-class.bat" org.apache.kafka.tools.ConsoleProducer %*
 EndLocal

checkstyle/import-control.xml

@@ -312,6 +312,8 @@
     <allow pkg="scala.collection" />
     <allow pkg="org.apache.kafka.coordinator.transaction" />
     <allow pkg="org.apache.kafka.coordinator.group" />
+    <allow pkg="org.apache.kafka.tools" />
+    <allow pkg="org.apache.kafka.tools.api" />
 
     <subpackage name="consumer">
       <allow pkg="org.apache.kafka.tools"/>

checkstyle/suppressions.xml

@@ -269,7 +269,7 @@
     <suppress checks="BooleanExpressionComplexity"
               files="(StreamsResetter|DefaultMessageFormatter).java"/>
     <suppress checks="NPathComplexity"
-              files="(DefaultMessageFormatter|ProducerPerformance|StreamsResetter|Agent|TransactionalMessageCopier|ReplicaVerificationTool).java"/>
+              files="(DefaultMessageFormatter|ProducerPerformance|StreamsResetter|Agent|TransactionalMessageCopier|ReplicaVerificationTool|LineMessageReader).java"/>
     <suppress checks="ImportControl"
               files="SignalLogger.java"/>
     <suppress checks="IllegalImport"

core/src/main/scala/kafka/utils/ToolsUtils.scala

@@ -20,20 +20,6 @@ import joptsimple.OptionParser
 import org.apache.kafka.server.util.CommandLineUtils
 
 object ToolsUtils {
-
-  def validatePortOrDie(parser: OptionParser, hostPort: String): Unit = {
-    val hostPorts: Array[String] = if (hostPort.contains(','))
-      hostPort.split(",")
-    else
-      Array(hostPort)
-    val validHostPort = hostPorts.filter { hostPortData =>
-      org.apache.kafka.common.utils.Utils.getPort(hostPortData) != null
-    }
-    val isValid = validHostPort.nonEmpty && validHostPort.length == hostPorts.length
-    if (!isValid)
-      CommandLineUtils.printUsageAndExit(parser, "Please provide valid host:port like host1:9091,host2:9092\n ")
-  }
-
   /**
    * This is a simple wrapper around `CommandLineUtils.printUsageAndExit`.
    * It is needed for tools migration (KAFKA-14525), as there is no Java equivalent for return type `Nothing`.