一、          节点故障或磁盘下线1.      Kafka节点服务异常通过FusionInsight Manager页面，选择“集群->Kafka”，查看当前Kafka集群当前状态，状态是否是良好；如果状态不是良好，说明Kafka服务异常。2.      磁盘下线通过FusionInsight Manager页面查看告警信息，是否有“Kafka数据目录状态异常”的告警，如果有，server.log日志分析告警产生原因，重启数据目录异常节点。二、          网络异常检查网络是否正常方法如下：使用ping -s 15000 IP命令查看是否有网络丢包，如果网络延迟大于5ms，说明网络延迟高。使用scp从客户端传输一个大文件到kafka服务侧节点，查看传输速率。一般，未完全同步的分区总数监控曲线一直在变化或者通过kafka-topics.sh  --zookeeper zk业务IP:24002/kafka --describe --under-replicated-partitions命令查询处理的未完全同步的分区一直在变化，网络原因的可能性比较大。三、          副本线程下线1.     查看未完全同步分区所在节点执行kafka-topics.sh  --zookeeper zk业务IP:24002/kafka --describe --under-replicated-partitions命令查看所有未完全同步的分区，对比Replicas和ISR列表中的节点，判断未同步的分区是否集中在一个或者几个节点上。例如，下图中ISR列表中都是3未同步。    2.     查看节点信息执行kafka-broker-info.sh  --zookeeper zk业务IP:24002/kafka查看未同步分区所在节点。3.     Kafka配置页面查看“num.replica.ers”的配置大小。4.     切omm用户打jstack信息根据第二步获取的节点IP，连上此节点，su - omm切omm用户，jps |grep Kafka获取Kafka进程的pid，执行“jstack -l kafkapid >>文件名”，然后cat jstack文件|grep ReplicaFetcherThread-查看同步线程。注意：一个副本同步线程的格式如：ReplicaFetcherThread-tid-bid，tid代表一个线程编号，如果配置多个（由服务端配置参数num.replica.ers决定），这个线程号会从0开始依次增加；bid表示leader所在节点的broker-id，如果两个broker之间存在主副本关系，比如：leader在broker-5上，follower在broker-6上，那么broker-6至少会有一个线程名为ReplicaFetcherThread-0-5；如果num.replica.ers配置为2，broker-6 可能会存在两个副本同步线程ReplicaFetcherThread-0-5，ReplicaFetcherThread-1-5。一个节点理论上能够出现的副本同步线程个数为num.replica.ers *（集群broker实例个数 - 1）。但并不是num.replica.ers配置多少就一定会出现这么多个副本同步线程，线程的数量会根据当前broker节点的分区总数来决定，如果节点上面的副本数很少，kafka会判定分区少而没有必要启动这么多副本同步线程，则副本同步线程个数可能少于num.replica.ers *（broker实例个数 - 1）。通常建议num.replica.ers值配置为3即可，太大会影响CPU的使用率）。5.     查看分区同步线程是否下线。如果未查询到同步线程，或者num.replica.ers配置为1且同步线程个数小于（broker实例个数 - 1）的值，或者没有查找此节点到未同步的leader节点的线程，则有同步线程下线。具体可根据未完全同步分区开始上升的时间去server.log日志中排查原因，可搜索“Stopped”、“Shutdown”等关键字查找“ReplicaFetcherThread-”开头的线程是否异常下线。6.     副本线程下线相关BUG1） java.nio.BufferOverflowException2） unexpected offset in append to3） Attempt to append to a full index4） Trying to roll a new log segment for partition以上BUG均在6518及之后版本解决。四、          节点异常退服节点退服前未将退服节点上的分区迁移至未退服节点，具体查看方式是执行kafka-topics.sh  --zookeeper zk业务IP:24002/kafka --describe --under-replicated-partitions命令查看所有未完全同步的分区，对比Replicas和ISR列表中的节点，判断未同步的分区是否集中在一个或者几个节点上，如果是，执行kafka-broker-info.sh  --zookeeper zk业务IP:24002/kafka查看节点是否在此列表中，如果没有，确认之前是否退服过节点，如果是，可以确定是退服节点导致。解决办法是迁移退服节点上的分区到正常节点。五、          Kafka节点压力大iostat -d -x 1查看CPU参数idle和IO参数util，idle值越高，CPU越空闲，util值越高，IO使用率越高。如果idle值小于20%，top -Hp kafkaPid查看CPU使用率高的线程，打jstack日志分析具体的原因；如果util值大于80%，查看磁盘对应的读写速率、await和svctm的大小判断对IO影响大的原因，如果读写速率比较大，排查kafka读写请求和延时，如果awati远大于svctm，IO队列太长，应用响应时间很慢。Kafka的数据流量监控中在未同步分区出现时有没有陡增，字节输入输出流量是否达到网络阈值。

我们使用kafka时，有时候会遇到发送数据失败的情况，其原因及解决方案如下：1.     Kafka topic leader为-1Kafka客户端执行如下命令查看topic的leader信息：kafka-topics.sh --describe --zookeeper zk业务IP:24002/kafka如果leader为-1，需查看Replicas中的副本节点是否正常，查看命令如下：kafka-broker-info.sh --zookeeper zk业务IP:24002/kafka命令中可以查询到brokerid，说明节点kafka服务正常如果leader为-1的分区对应的Replicas中节点都不正常，需要先恢复异常节点kafka服务。如果都正常但ISR列表中无节点信息，或者ISR列表中的节点不正常，需要查看Kafka服务端的unclean.leader.election.enable参数是否为true或者topic端是否配置此参数为true。如果都不为true，需要对此topic修改配置为true。Kafka服务端的unclean.leader.election.enable参数配置查看方式如下：                                             Kafka topic端unclean.leader.election.enable参数配置查看方式如下：kafka-topics.sh --describe --zookeeper zk业务IP:24002/kafka --topic topicName如果Config中无unclean.leader.election.enable信息，则与服务端配置一致。如果有，则此配置优先级高于服务端配置。修改topic端此配置的方式如下：kafka-topics.sh --alter --topic topicName --zookeeper zk业务IP:24002/kafka --config unclean.leader.election.enable=true2.     DNS配置导致执行 vi /etc/resolv.conf，如果有“nameserver X.X.X.X”,把此内容注释掉3.     网络异常生产端节点ping服务端IP，ping -s 15000 IP，如果延迟高于5ms，说明网络延迟过高，也可通过长ping来判断是否有网络丢包。4.     CPU或者IO过高也可能导致连接失败iostat -d -x 1查看CPU参数idle和IO参数util，idle值越高，CPU越空闲，util值越高，IO使用率越高。如果idle值小于20%，top -Hp kafkaPid查看CPU使用率高的线程，打jstack日志分析具体的原因；如果util值大于80%，查看磁盘对应的读写速率、await和svctm的大小判断对IO影响大的原因，如果读写速率比较大，排查kafka读写请求和延时，如果awati远大于svctm，IO队列太长，应用响应时间很慢。5.     磁盘坏道或者其他原因也可能导致连接失败如果server.log日志中有大量“java.io.IOException: Connection to IP:21007 (id: 2 rack: null) failed”日志，查看IP节点操作系统日志中是否有“Sense Key : Medium Error”信息，如果有，说明出现磁盘坏道，需修复或更换磁盘。另外，还需要打此节点的jstack信息，排查是否有阻塞或者死锁。如果是C80版本，且jstack信息中有如下信息，需打死锁补丁：6.     如果报“TimeoutException”或偶尔发送失败，可先调大request.timeout.ms，并查看服务端num.io.threads和num.network.threads是否可以优化（这两个参数一般调整为节点磁盘个数的倍数）。根据发送的数据量的大小也可适当调整batch.size、buffer.memory、linger.ms的大小。如果发送的数据量很大且可容忍一定时延，也可以考虑开启压缩，compression.type指定压缩方式，可配置为“gzip”、“snappy”或“lz4”。7.     ssh卡住ssh -v -p 端口号 异常节点ip如果如上图所示卡住，解决方式是将GSSAPIAuthentication修改为no8.     如果是集群外客户端生产发送失败，还可以通过集群内客户端测试下生产是否成功，进一步减小排查方向。

【操作步骤&问题现象】kafka节点扩容后执行重分布，如果节点存在系统盘慢盘是否有影响？

【功能模块】ranger给kafka指定权限策略【操作步骤&问题现象】1、创建新用户，不添加角色和用户组，该用户可以访问kafka所有topic2、给用户添加角色或者配置用户组，以及添加相应的权限策略后，权限策略不生效【截图信息】问题一：创建一个新用户，正常应该是没有权限访问kafka的topic，但是却可以访问问题二：使用ranger添加禁止策略后还是策略不生效，还是可以访问kafka的topic【也是用rangeradmin用户添加过，依然不行，云服务器线下安装的FusionInsight 版本号8.0.2】添加的策略【日志信息】（可选，上传日志内容或者附件）无日志信息

【功能模块】    ranger给kafka指定权限策略【操作步骤&问题现象】1、创建新用户，不添加角色和用户组，该用户可以访问kafka所有topic2、给用户添加角色或者配置用户组，以及添加相应的权限策略后，权限策略不生效【截图信息】    问题1：问题2：添加的策略【日志信息】（可选，上传日志内容或者附件）    无日志信息

                我们使用kafka时，有时候会遇到发送数据失败的情况，其原因及解决方案如下：1.     Kafka topic leader为-1Kafka客户端执行如下命令查看topic的leader信息：kafka-topics.sh --describe --zookeeper zk业务IP:24002/kafka如果leader为-1，需查看Replicas中的副本节点是否正常，查看命令如下：kafka-broker-info.sh --zookeeper zk业务IP:24002/kafka命令中可以查询到brokerid，说明节点kafka服务正常如果leader为-1的分区对应的Replicas中节点都不正常，需要先恢复异常节点kafka服务。如果都正常但ISR列表中无节点信息，或者ISR列表中的节点不正常，需要查看Kafka服务端的unclean.leader.election.enable参数是否为true或者topic端是否配置此参数为true。如果都不为true，需要对此topic修改配置为true。Kafka服务端的unclean.leader.election.enable参数配置查看方式如下：                                             Kafka topic端unclean.leader.election.enable参数配置查看方式如下：kafka-topics.sh --describe --zookeeper zk业务IP:24002/kafka --topic topicName如果Config中无unclean.leader.election.enable信息，则与服务端配置一致。如果有，则此配置优先级高于服务端配置。修改topic端此配置的方式如下：kafka-topics.sh --alter --topic topicName --zookeeper zk业务IP:24002/kafka --config unclean.leader.election.enable=true2.     DNS配置导致执行 vi /etc/resolv.conf，如果有“nameserver X.X.X.X”,把此内容注释掉3.     网络异常生产端节点ping服务端IP，ping -s 15000 IP，如果延迟高于5ms，说明网络延迟过高，也可通过长ping来判断是否有网络丢包。4.     CPU或者IO过高也可能导致连接失败iostat -d -x 1查看CPU参数idle和IO参数util，idle值越高，CPU越空闲，util值越高，IO使用率越高。如果idle值小于20%，top -Hp kafkaPid查看CPU使用率高的线程，打jstack日志分析具体的原因；如果util值大于80%，查看磁盘对应的读写速率、await和svctm的大小判断对IO影响大的原因，如果读写速率比较大，排查kafka读写请求和延时，如果awati远大于svctm，IO队列太长，应用响应时间很慢。5.     磁盘坏道或者其他原因也可能导致连接失败如果server.log日志中有大量“java.io.IOException: Connection to IP:21007 (id: 2 rack: null) failed”日志，查看IP节点操作系统日志中是否有“Sense Key : Medium Error”信息，如果有，说明出现磁盘坏道，需修复或更换磁盘。另外，还需要打此节点的jstack信息，排查是否有阻塞或者死锁。如果是C80版本，且jstack信息中有如下信息，需打死锁补丁：6.     如果报“TimeoutException”或偶尔发送失败，可先调大request.timeout.ms，并查看服务端num.io.threads和num.network.threads是否可以优化（这两个参数一般调整为节点磁盘个数的倍数）。根据发送的数据量的大小也可适当调整batch.size、buffer.memory、linger.ms的大小。如果发送的数据量很大且可容忍一定时延，也可以考虑开启压缩，compression.type指定压缩方式，可配置为“gzip”、“snappy”或“lz4”。7.     ssh卡住ssh -v -p 端口号 异常节点ip如果如上图所示卡住，解决方式是将GSSAPIAuthentication修改为no8.     如果是集群外客户端生产发送失败，还可以通过集群内客户端测试下生产是否成功，进一步减小排查方向。

在使用Kafka时，我们有时候会遇到生产时连接Kafka失败的情况，原因及解决办法如下：1.       Kafka服务异常通过FusionInsight Manager页面，选择“集群->Kafka”，查看当前Kafka集群当前状态，状态是否是良好；如果状态不是良好，说明Kafka服务异常。2.       生产命令错误如果使用21005端口，命令如下：kafka-console-producer.sh --broker-list Kafka业务IP:21005 --topic topicName如果使用21007端口，命令如下：kafka-console-producer.sh --broker-list Kafka业务IP:21005 --topic topicName -- --producer.config ../config/producer.properties其中，producer.properties中的security.protocol必须是SASL_PLAINTEXT。如果使用21007端口，security.protocol必须是SASL_PLAINTEXT，Kafka服务端配置ssl.mode.enable必须是true。3.       无权限如果使用21005端口生产，查看“allow.everyone.if.no.acl.found”参数是否为“ture”，如果不是true，修改为true后保存重启。如果使用21007端口生产，如果连接提示中有“TOPIC_AUTHORIZATION_FAILED”或“Not authorized to access topic”信息，说明无消费的topic的权限。执行klist查看用户名，执行id -Gn 用户名，确定该用户名的属组，如果id -Gn命令执行失败，可能sssd服务异常，可使用“systemctl start sssd”重启进程。如果属组中无kafkaadmin或supergroup或kafka，说明用户没有配置kafka角色，需添加kafka角色信息；如果只有kafka，需要查看该用户是否有topic的生产权限，查看命令如下：kafka-acls.sh --authorizer-properties zookeeper.connect=zk业务IP:24002/kafka --list --topic topicName如果没有配置，需添加此用户权限信息或此用户加入kafkaadmin组。添加生产权限信息命令如下：kafka-acls.sh --authorizer-properties zookeeper.connect= zk业务IP:24002/kafka --topic topicName --producer --add --allow-principal User:用户名4.       Kafka配置中version配置错误通过FusionInsight Manager页面，选择“集群->Kafka->配置->全部配置”中搜“version”，查看如下version配置：   651X版本对应的是1.1-IV0，C80对应版本是0.11.0-IV2，C70对应版本是0.10-IV1。如果配置不对，修改为对应配置后保存并重启Kafka所有实例。5.       Kafka端口不可用执行netstat -anp|grep 端口号查看端口是否有监听或者连接，如果有，在生产的客户端执行ssh -v -p 21005 IP，查看连接是否成功，如果报“Connection timeout”说明客户端不可访问服务端的端口，服务端需对客户端开放访问端口权限。6.       网络异常检查网络是否正常方法如下：Ø  使用ping命令查看是否有网络丢包，ping -s 15000 IP，如果网络延迟大于5ms，说明网络延迟高。Ø  使用scp从客户端传输一个大文件到kafka服务侧节点，查看传输速率。7.       如果是集群外客户端连接失败，还可以通过集群内客户端测试下生产是否成功，进一步减小排查方向。

一、          收集Kafka实例和分区信息1.       执行kafka-broker-info.sh --zookeeper zk业务IP:24002/kafka 命令获取Kafka实例信息2.       执行kafka-topics.sh --describe --zookeeper zk业务IP:24002/kafka --unavailable-partitions 命令获取Kafka未同步分区3.       对比Replicas和ISR列表中的信息，确定哪些节点的分区未同步，确定未同步的节点正常与否。例如，如下案例中，对比Replicas和ISR列表中的信息发现3节点没有同步，而正常的kafka实例中没有3。                      二、          准备迁移方案根据步骤一中的未完全同步的分区即需要迁移的分区，把需要迁移的所有分区写入一个新的json文件，文件中replicas中的信息需要根据步骤一中第2步查询结果中的Replicas把退服节点的brokerid改为正常节点的brokerid(即步骤一中第1步查询到的brokerid)，其他不变，格式如下所示：{"version":1,"partitions":[{"topic":"test1","partition":0"replicas":[1,2]},{"topic":"test1","partition":1"replicas":[1,2]},{"topic":"test1","partition":5,"replicas":[2,1]},{"topic":"test1","partition":3,"replicas":[2,1]}]}如果涉及的分区比较多，也可按如下方式生成迁移的json文件。1.       根据步骤一中未完全同步的分区中的topic信息编写迁移topic的json文件，例如json文件名为generate.json，格式如下：{"topics":[{"topic": "test1"},{"topic": "test2"}], "version":1}2.       执行命令kafka-reassign-partitions.sh --zookeeper zk业务IP:24002/kafka --topics-to-move-json-file generate.json文件路径  --broker-list 1,2 --generate 生成迁移方案，其中，broker-list中对应参数是所有正常的broker实例的id，即步骤一中1获取的所有brokerid，不同brokerid间用逗号隔开。3.       把上一步得到的“Proposed partition reassignment configuration”的json文件写入迁移的json文件中，例如move.json，此json文件中只保留需要迁移的topic和partition的信息。如下图所示，红色方框内即迁移的json文件信息。三、          执行迁移执行kafka-reassign-partitions.sh --zookeeper zk业务IP:24002/kafka  --reassignment-json-file move.json文件路径 --execute 执行迁移四、          查看迁移进度执行kafka-reassign-partitions.sh --zookeeper zk业务IP:24002/kafka  --reassignment-json-file move.json文件路径 --verify 查看迁移进度，当所有partition都提示completed successfully时说明迁移完成。五、          查看是否所有分区都同步执行kafka-topics.sh --describe --zookeeper zk业务IP:24002/kafka --unavailable-partitions 命令获取Kafka未同步分区，如果结果为空，说明所有分区都同步，操作完成；如果有leader为-1，按步骤六执行。六、          Leader为-1的解决办法1.      Kafka配置中查看unclean.leader.election.enable配置，如果配置为true，只执行下面步骤2即可；如果配置为false，执行下面步骤3、4、5。2.      切controller：进zookeeper客户端，执行 zkCli.sh -server zk业务IP：24002/kafka进入zk，执行deleteall /controller 切controller。3.      Kafka客户端修改参数配置：kafka-topics.sh --alter --topic topicName --zookeeper zk业务IP:24002/kafka --config unclean.leader.election.enable=true，其中，topicName是leader为-1的topic。4.      按步骤2切controller。5.      查看leader为-1的分区leader是否正常，正常后在kafka客户端执行kafka-topics.sh --alter --topic topicName --zookeeper 100.112.22.233:24002/kafka --delete-config unclean.leader.election.enable。

配置好了，启动就报错Error: VM option 'UseG1GC' is experimental and must be enabled via -XX:+UnlockExperimentalVMOptions.Error: Could not create the Java Virtual Machine.Error: A fatal exception has occurred. Program will exit.求解？？？

【功能模块】kafka【操作步骤&问题现象】flink作业连接kafka【截图信息】【日志信息】（可选，上传日志内容或者附件）内部联系 30000836

【功能模块】bin/flink run --class com.huawei.flink.example.sqljoin.WriteIntoKafka4SQLJoin /opt/Flink_test/flink-examples-1.0.jar --topic topic-test --bootstrap.servers xxx.xxx.xxx.xxx:21005运行测试 例子   com.huawei.flink.example.sqljoin.WriteIntoKafka4SQLJoin运行不通过  【操作步骤&问题现象】1、2、【截图信息】【日志信息】（可选，上传日志内容或者附件）--sasl.kerberos.service.name kafka --ssl.truststore.location /home/truststore.jks --ssl.truststore.password huawei******************************************************************************************<topic> is the kafka topic name<bootstrap.servers> is the ip:port list of brokers******************************************************************************************java.lang.NoClassDefFoundError: org/apache/kafka/clients/producer/Callback        at com.huawei.flink.example.sqljoin.WriteIntoKafka4SQLJoin.main(WriteIntoKafka4SQLJoin.java:53)        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)        at java.lang.reflect.Method.invoke(Method.java:498)        at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:529)        at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:421)        at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:430)        at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:814)        at org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:288)        at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:213)        at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1051)        at org.apache.flink.client.cli.CliFrontend.lambda$main$11(CliFrontend.java:1127)        at java.security.AccessController.doPrivileged(Native Method)        at javax.security.auth.Subject.doAs(Subject.java:422)        at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1729)        at org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)        at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1127)Caused by: java.lang.ClassNotFoundException: org.apache.kafka.clients.producer.Callback        at java.net.URLClassLoader.findClass(URLClassLoader.java:382)        at java.lang.ClassLoader.loadClass(ClassLoader.java:424)        at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)        at java.lang.ClassLoader.loadClass(ClassLoader.java:357)        ... 18 more

## 背景 在搭建Kafka集群时，通常会按照机架部署机器。Kafka在新建topic时，为了提高集群的可靠性， 会将同一分区的不同副本分布在不同的机架内，这样即使有一个机架上的机器宕掉，也不会影响服务的可用性和稳定性 如下图：集群中有三个机架，每个机架内部3个节点，在这个集群内创建了3分区3副本的topic  ## 生产/消费模型  - 生产模型 - 各个Producer客户端会连接对应分区的 **leader** 进行数据生产, 对应分区的follower会定期从leader处同步数据 - 消费模型 - 各个Consumer客户端连接对应分区的 **leader** 进行数据消费 ## 上述生产/消费模型弊端 - 生产和消费的客户端对接的全部都是**leader**分区，当客户端较多时，会给集群造成较大压力 - 同一个分区的follower副本，仅作为数据备份，不对外提供服务 - 当跨机架间消费客户端增多时，会产生大量的机架间网络传输，给网络造成压力 ## 就近消费特性  - 打破follower副本不能对外提供服务的禁锢，允许消费者从follower副本处拉取数据, 优点如下： - 从follower副本消费数据，可大大降低leader副本节点压力 - 就近从同一机架内节点内消费数据，可大大降低集群内机架间网络数据传输量 - 降低客户端消费延迟 ## 如何开启就近消费特性 - 客户端配置 - 启动服务时，replica.selector.class = org.apache.kafka.common.replica.RackAwareReplicaSelector - 客户端配置 - 启动Consumer客户端时，配置 client.rack = /xxxx/xxxx

操作场景Flume采集 kafka 内容导入到 hbase前提条件已创建启用Kerberos认证的流集群。已在日志生成节点安装Flume客户端，请参见安装Flume客户端。已配置网络，使日志生成节点与流集群互通。操作步骤(1)  从HDFS客户端拷贝配置文件core-site.xml，hdfs-site.xml到Flume Client的配置目录 " /opt/FlumeClient/fusioninsight-flume-1.6.0/conf" 下。通常可以在分析集群Master节点HDFS客户端安装目录如“/opt/client/HDFS/hadoop/etc/hadoop/”下找到core-site.xml，hdfs-site.xml文件。(2)  从MRS集群下载用户的认证凭据。①在MRS Manager，单击“系统设置”。 ②在“权限配置”区域，单击“用户管理”。 ③在用户列表中选择需要的用户，单击后面的“更多”下载用户凭据。 ④解压下载的用户凭据文件，获取krb5.conf和user.keytab文件。(3) 将上一步获得的krb5.conf和user.keytab拷贝到Flume Client所在节点的配置目录 " /opt/FlumeClient/fusioninsight-flume-1.X.X/conf" 下。（4）修改配置文件jaas.conf，        ① 将/opt/Bigdata/MRS_x.x.x/1_x_Flume/etc/" 的jaas.conf文件拷贝到/opt/FlumeClient/fusioninsight-flume-1.X.XS/conf下②  修改jaas.conf文件中的参数principal 与keyTabPrincipal: 可以kinit 用户名进行认证查看参数keyTab: 定义的用户认证文件完整路径即步骤(1)中保存user.keytab认证文件的目录 (5) 修改配置文件flume-env.sh，文件所在目录 /opt/FlumeClient/fusioninsight-flume-1.6.0/conf。在 “-XX:+UseCMSCompactAtFullCollection”后面，增加以下内容：-Djava.security.krb5.conf=/opt/FlumeClient/fusioninsight-flume-1.6.0/conf /krb5.conf -Djava.security.auth.login.config=/opt/FlumeClient/fusioninsight-flume-1.6.0/conf /jaas.conf -Dzookeeper.request.timeout=120000krb5.conf与jaas.conf根据实际情况，修改配置文件路径，然后保存并退出配置文件。（6）假设Flume客户端安装路径为“/opt/FlumeClient”，执行以下命令，重启Flume客户端：cd /opt/FlumeClient/fusioninsight-flume-1.6.0/bin./flume-manage.sh restart（7）执行以下命令，修改Flume客户端配置文件“properties.properties”。vi Flume客户端安装目录/fusioninsight-flume-1.6.0/conf/properties.properties将以下内容保存到文件中：client.sources = obsclient.channels = flumeclient.sinks = hdfsclient.sources.obs.type = org.apache.flume.source.kafka.KafkaSourceclient.sources.obs.monTime = 0client.sources.obs.nodatatime = 0client.sources.obs.batchSize = 1client.sources.obs.batchDurationMillis = 1client.sources.obs.keepTopicInHeader = falseclient.sources.obs.keepPartitionInHeader = falseclient.sources.obs.kafka.bootstrap.servers =xxxxxxxx:21007client.sources.obs.kafka.consumer.group.id = consumer_groupclient.sources.obs.kafka.topics = topic01client.sources.obs.kafka.security.protocol = SASL_PLAINTEXTclient.sources.obs.kafka.kerberos.domain.name = hadoop.xxx.comclient.sources.obs.channels = flumeclient.channels.flume.type = memoryclient.channels.flume.capacity = 10000client.channels.flume.transactionCapacity = 1000client.channels.flume.channelfullcount = 10client.channels.flume.keep-alive = 3client.channels.flume.byteCapacity =client.channels.flume.byteCapacityBufferPercentage = 20client.sinks.hdfs.type = hdfsclient.sinks.hdfs.hdfs.path = hdfs://hacluster/tmp/ldpclient.sinks.hdfs.montime =client.sinks.hdfs.hdfs.filePrefix = over_%{basename}client.sinks.hdfs.hdfs.fileSuffix =client.sinks.hdfs.hdfs.inUsePrefix =client.sinks.hdfs.hdfs.inUseSuffix = .tmpclient.sinks.hdfs.hdfs.idleTimeout = 0client.sinks.hdfs.hdfs.inUseSuffix = .tmpclient.sinks.hdfs.hdfs.idleTimeout = 0client.sinks.hdfs.hdfs.batchSize = 1000client.sinks.hdfs.hdfs.codeC =client.sinks.hdfs.hdfs.fileType = DataStreamclient.sinks.hdfs.hdfs.maxOpenFiles = 5000client.sinks.hdfs.hdfs.writeFormat = Writableclient.sinks.hdfs.hdfs.callTimeout = 10000client.sinks.hdfs.hdfs.threadsPoolSize = 10client.sinks.hdfs.hdfs.rollTimerPoolSize = 1client.sinks.hdfs.hdfs.kerberosPrincipal = ldpclient.sinks.hdfs.hdfs.kerberosKeytab = /opt/FlumeClient/fusioninsight-flume-1.6.0/conf/user.keytabclient.sinks.hdfs.hdfs.round = falseclient.sinks.hdfs.hdfs.roundUnit = secondclient.sinks.hdfs.hdfs.useLocalTimeStamp = trueclient.sinks.hdfs.hdfs.failcount = 10client.sinks.hdfs.hdfs.fileCloseByEndEvent = falseclient.sinks.hdfs.hdfs.rollInterval = 30client.sinks.hdfs.hdfs.rollSize = 1024client.sinks.hdfs.hdfs.rollCount = 10client.sinks.hdfs.hdfs.batchCallTimeout = 0client.sinks.hdfs.serializer.appendNewline = trueclient.sinks.hdfs.channel = flume----------------------------------------------------------------client.sources.obs.kafka.kerberos.domain.name  配置可以从hosts文件中找到hadoop开头的

各位大佬好，有没有知道kafka如何释放，删除？我的删除功能是灰色的不能用

【功能模块】kafka【操作步骤&问题现象】参考apache kylin对接文档，做sample-streaming.sh用例的时候在kylin.log遇到报错【截图信息】【日志信息】（可选，上传日志内容或者附件）2020-12-16 15:23:18,010 ERROR [http-nio-7070-exec-5] controller.CubeController:434 : Timeout expired while fetching topic metadataorg.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata2020-12-16 15:23:18,014 ERROR [http-nio-7070-exec-5] controller.BasicController:63 :org.apache.kylin.rest.exception.InternalErrorException: Timeout expired while fetching topic metadata        at org.apache.kylin.rest.controller.CubeController.buildInternal(CubeController.java:435)        at org.apache.kylin.rest.controller.CubeController.rebuild2(CubeController.java:418)