kafka开启ACL,请求慢出现超时的解决办法

Posted 2022-07-12 不识君的荒漠

问题描述

正常情况下，使用某个kafka管理平台（或者是使用kafka提供的admin client）调用kafka的接口查询相关信息，比如查询topic列表，请求是很快的，如下：

正常情况（网络也没问题），请求一般是很快的，如果通过一些管理平台，页面显示也是很流畅的。

目前有几次遇到请求特别慢，接近10秒甚至数十秒的情况，一些管理平台就可能出现超时现象或者查询的特别慢。

我目前遇到2次，说明如下：

场景1

环境信息如下：

• 本机操作系统：mac OS
• kafka部署环境：在国内的某个云虚拟机上部署
• 开启VPN：是，本机电脑开启了VPN
• 开启ACL：是，kafka集群开启ACL（SASL）

当时在公司的时候是没有问题的，回家的时候因为要开启VPN就会出现这个问题，管理平台页面特别慢，甚至查询不出来。

场景2

环境信息如下：

• 本机操作系统：win10
• kafka部署环境：virtualbox安装的虚拟机，操作系统是centos
• 开启ACL：是，kafka集群开启ACL（SASL）

如果不开启ACL是没有问题的。

解决办法

因为随便试了下，本机这个问题现在消失了，暂时复现不出来，也没法debug了，所以根本原因还没找到。

下面提供一些办法进行尝试，这个操作是针对客户端所在的主机上（比如管理平台部署的主机）。

刷新hosts文件

我是win10系统，hosts文件位置：C:\\Windows\\System32\\drivers\\etc\\hosts

将这个文件copy到桌面，编辑增加一条配置之后，再copy到C:\\Windows\\System32\\drivers\\etc目录下覆盖。

然后我本机的kafka连接慢的问题就解决了~~~

IP映射

还是修改hosts文件，增加一个kafka主机的IP映射：

192.168.56.101		centos-mini

连接kafka的时候使用后面这个域名。

p.s. 该方法我并未来得及验证，因为更新hosts文件的时候已经解决了，这个方法是猜测的，如果是dns解析导致的变慢，那这个方法是可以解决的。

缓存连接

这是我最初考虑的解决方法，因为每次连接的时候sasl验证方式的实现，在客户端连接时有一个握手，对身份进行校验（普通的明文协议无认证的代码实现中，校验这一块是空实现也就是不验证，所以问题应该在这个范围）。

因此请求慢主要是客户端的连接慢（不是TCP连接慢）。因此考虑把连接缓存起来，不要每次请求建立即可，所以这个我是修改kafka管理平台的代码直接缓存连接，效果也是非常好，只有第一次请求的时候比较慢，示例如下（增加了个开关，如果缓存连接，直接从缓存获取连接，基于guava的cache进行实现）：

    protected def withAdminClient(f: Admin => Any): Any = 

        val admin = if (config.isCacheAdminConnection())  AdminCache.cache.get(ContextConfigHolder.CONTEXT_CONFIG.get().getBootstrapServer()) else createAdminClient()
        try 
            f(admin)
         finally 
            if (!config.isCacheAdminConnection) 
                admin.close()
            
        
    

object AdminCache 

    private val log: Logger = LoggerFactory.getLogger(this.getClass)

    private val cacheLoader = new CacheLoader[String, Admin] 
        override def load(key: String): Admin = KafkaConsole.createAdminClient()
    

    private val removeListener = new RemovalListener[String, Admin] 
        override def onRemoval(notification: RemovalNotification[String, Admin]): Unit = 
            Future 
                log.warn("Close expired admin connection: ", notification.getKey)
                notification.getValue.close()
                log.warn("Close expired admin connection complete: ", notification.getKey)
            (KafkaConsole.ec)

        
    
    val cache = new TimeBasedCache[String, Admin](cacheLoader, removeListener)

消费者和生产的连接缓存实现也类似，不再说明。

这三种方法，可以根据情况进行尝试，因为可能主机在运维或其它部门管控等原因，自己无法操作主机；或者是客户端侧的代码自己不方便修改。现实情况总是很复杂的，所以这里只是一个解决建议。