升级kafka版本从2.8.0 -> 3.2.0，增加DockerCompose部署说明

2022-06-30 20:11:29 +08:00
parent d062e18940
commit b163e5f776
6 changed files with 77 additions and 57 deletions
--- a/src/main/java/com/xuxd/kafka/console/beans/AclEntry.java
+++ b/src/main/java/com/xuxd/kafka/console/beans/AclEntry.java
@@ -1,18 +1,15 @@
 package com.xuxd.kafka.console.beans;

-import java.util.Objects;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.kafka.common.acl.AccessControlEntry;
-import org.apache.kafka.common.acl.AccessControlEntryFilter;
-import org.apache.kafka.common.acl.AclBinding;
-import org.apache.kafka.common.acl.AclBindingFilter;
-import org.apache.kafka.common.acl.AclOperation;
-import org.apache.kafka.common.acl.AclPermissionType;
+import org.apache.kafka.common.acl.*;
 import org.apache.kafka.common.resource.PatternType;
 import org.apache.kafka.common.resource.ResourcePattern;
 import org.apache.kafka.common.resource.ResourcePatternFilter;
 import org.apache.kafka.common.resource.ResourceType;
 import org.apache.kafka.common.security.auth.KafkaPrincipal;
+import org.apache.kafka.common.utils.SecurityUtils;
+
+import java.util.Objects;

 /**
 * kafka-console-ui.
@@ -41,7 +38,9 @@ public class AclEntry {
        entry.setResourceType(binding.pattern().resourceType().name());
        entry.setName(binding.pattern().name());
        entry.setPatternType(binding.pattern().patternType().name());
-        entry.setPrincipal(KafkaPrincipal.fromString(binding.entry().principal()).getName());
+//        entry.setPrincipal(KafkaPrincipal.fromString(binding.entry().principal()).getName());
+        // 3.x版本使用该方法
+        entry.setPrincipal(SecurityUtils.parseKafkaPrincipal(binding.entry().principal()).getName());
        entry.setHost(binding.entry().host());
        entry.setOperation(binding.entry().operation().name());
        entry.setPermissionType(binding.entry().permissionType().name());
--- a/src/main/scala/kafka/console/BrokerApiVersion.scala
+++ b/src/main/scala/kafka/console/BrokerApiVersion.scala
@@ -18,6 +18,7 @@ import org.apache.kafka.common.network.Selector
 import org.apache.kafka.common.protocol.Errors
 import org.apache.kafka.common.requests._
 import org.apache.kafka.common.utils.{KafkaThread, LogContext, Time}
+import org.slf4j.{Logger, LoggerFactory}

 import java.io.IOException
 import java.util.Properties
@@ -34,7 +35,9 @@ import scala.util.{Failure, Success, Try}
 * @author xuxd
 * @date 2022-01-22 15:15:57
 * */
-object BrokerApiVersion extends Logging {
+object BrokerApiVersion{
+
+  protected lazy val log : Logger = LoggerFactory.getLogger(this.getClass)

    def listAllBrokerApiVersionInfo(): java.util.HashMap[Node, NodeApiVersions] = {
        val res = new java.util.HashMap[Node, NodeApiVersions]()
@@ -48,7 +51,7 @@ object BrokerApiVersion extends Logging {
                        case Success(v) => {
                            res.put(broker, v)
                        }
-                        case Failure(v) => logger.error(s"${broker} -> ERROR: ${v}\n")
+                        case Failure(v) => log.error(s"${broker} -> ERROR: ${v}\n")
                    }
            }
        } finally {
@@ -149,12 +152,12 @@ object BrokerApiVersion extends Logging {
            val response = sendAnyNode(request).asInstanceOf[MetadataResponse]
            val errors = response.errors
            if (!errors.isEmpty) {
-                logger.info(s"Metadata request contained errors: $errors")
+                log.info(s"Metadata request contained errors: $errors")
            }

            // 在3.x版本中这个方法是buildCluster 代替cluster()了
-            //            response.buildCluster.nodes.asScala.toList
-            response.cluster().nodes.asScala.toList
+                        response.buildCluster.nodes.asScala.toList
+//            response.cluster().nodes.asScala.toList
        }

        def listAllBrokerVersionInfo(): Map[Node, Try[NodeApiVersions]] =
@@ -277,40 +280,40 @@ object BrokerApiVersion extends Logging {

            // 版本不一样，这个地方的兼容性问题也不一样了
            // 3.x版本用这个
-            //            val networkClient = new NetworkClient(
-            //                selector,
-            //                metadata,
-            //                clientId,
-            //                DefaultMaxInFlightRequestsPerConnection,
-            //                DefaultReconnectBackoffMs,
-            //                DefaultReconnectBackoffMax,
-            //                DefaultSendBufferBytes,
-            //                DefaultReceiveBufferBytes,
-            //                requestTimeoutMs,
-            //                connectionSetupTimeoutMs,
-            //                connectionSetupTimeoutMaxMs,
-            //                time,
-            //                true,
-            //                new ApiVersions,
-            //                logContext)
+                        val networkClient = new NetworkClient(
+                            selector,
+                            metadata,
+                            clientId,
+                            DefaultMaxInFlightRequestsPerConnection,
+                            DefaultReconnectBackoffMs,
+                            DefaultReconnectBackoffMax,
+                            DefaultSendBufferBytes,
+                            DefaultReceiveBufferBytes,
+                            requestTimeoutMs,
+                            connectionSetupTimeoutMs,
+                            connectionSetupTimeoutMaxMs,
+                            time,
+                            true,
+                            new ApiVersions,
+                            logContext)

-            val networkClient = new NetworkClient(
-                selector,
-                metadata,
-                clientId,
-                DefaultMaxInFlightRequestsPerConnection,
-                DefaultReconnectBackoffMs,
-                DefaultReconnectBackoffMax,
-                DefaultSendBufferBytes,
-                DefaultReceiveBufferBytes,
-                requestTimeoutMs,
-                connectionSetupTimeoutMs,
-                connectionSetupTimeoutMaxMs,
-                ClientDnsLookup.USE_ALL_DNS_IPS,
-                time,
-                true,
-                new ApiVersions,
-                logContext)
+//            val networkClient = new NetworkClient(
+//                selector,
+//                metadata,
+//                clientId,
+//                DefaultMaxInFlightRequestsPerConnection,
+//                DefaultReconnectBackoffMs,
+//                DefaultReconnectBackoffMax,
+//                DefaultSendBufferBytes,
+//                DefaultReceiveBufferBytes,
+//                requestTimeoutMs,
+//                connectionSetupTimeoutMs,
+//                connectionSetupTimeoutMaxMs,
+//                ClientDnsLookup.USE_ALL_DNS_IPS,
+//                time,
+//                true,
+//                new ApiVersions,
+//                logContext)

            val highLevelClient = new ConsumerNetworkClient(
                logContext,
--- a/src/main/scala/kafka/console/KafkaConsole.scala
+++ b/src/main/scala/kafka/console/KafkaConsole.scala
@@ -91,14 +91,17 @@ class KafkaConsole(config: KafkaConfig) {
        }
    }

+    @Deprecated
    protected def withZKClient(f: AdminZkClient => Any): Any = {
-        val zkClient = KafkaZkClient(config.getZookeeperAddr, false, 30000, 30000, Int.MaxValue, Time.SYSTEM)
-        val adminZkClient = new AdminZkClient(zkClient)
-        try {
-            f(adminZkClient)
-        } finally {
-            zkClient.close()
-        }
+//        val zkClient = KafkaZkClient(config.getZookeeperAddr, false, 30000, 30000, Int.MaxValue, Time.SYSTEM)
+        // 3.x
+//        val zkClient = KafkaZkClient(config.getZookeeperAddr, false, 30000, 30000, Int.MaxValue, Time.SYSTEM, new ZKClientConfig(), "KafkaZkClient")
+//        val adminZkClient = new AdminZkClient(zkClient)
+//        try {
+//            f(adminZkClient)
+//        } finally {
+//            zkClient.close()
+//        }
    }

    protected def createAdminClient(props: Properties): Admin = {
--- a/src/main/scala/kafka/console/MessageConsole.scala
+++ b/src/main/scala/kafka/console/MessageConsole.scala
@@ -127,7 +127,7 @@ class MessageConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaConf
                                                record.offset(),
                                                record.timestamp(),
                                                record.timestampType(),
-                                                record.checksum(),
+//                                                record.checksum(),
                                                record.serializedKeySize(),
                                                record.serializedValueSize(),
                                                record.key(),