如何创建，删除，更改kafka的topic及属性

我们都知道，有时候我们创建topic的时候一些属性没有设置好，导致后面出现很多状况，下面介绍如何创建，修改，删除kafka主题的简单操作。如下介绍2种方法。
方法1：
首先，创建一个pojo类：
package com.ky.common;

/**

• @Author: xwj

• @Date: 2019/1/7 0007 13:53

• @Version 1.0
  */
  public class KafkaTopicBean {

  private String topicName;
  private Integer partition;
  private Integer replication;
  private String descrbe;

  String getTopicName() {
  return topicName;
  }

  public void setTopicName(String topicName) {
  this.topicName = topicName;
  }

  Integer getPartition() {
  return partition;
  }

  public void setPartition(Integer partition) {
  this.partition = partition;
  }

  Integer getReplication() {
  return replication;
  }

  public void setReplication(Integer replication) {
  this.replication = replication;
  }

  public String getDescrbe() {
  return descrbe;
  }

  public void setDescrbe(String descrbe) {
  this.descrbe = descrbe;
  }

  @Override
  public String toString() {
  return "KafkaTopicBean [topicName=" + topicName + ", partition=" + partition
  + ", replication=" + replication + ", descrbe=" + descrbe +"]";
  }

}
创建kafkaUtil工具类
package com.ky.common;

import kafka.admin.AdminUtils;
import kafka.admin.RackAwareMode;
import kafka.server.ConfigType;
import kafka.utils.ZkUtils;
import org.apache.kafka.common.security.JaasUtils;
import scala.collection.JavaConversions;

import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

/**

• @Author: xwj

• @Date: 2019/1/7 0007 13:54

• @Version 1.0
  */
  public class KafkaUtil {

  public static void createKafaTopic(String zkStr, KafkaTopicBean topic) {
  ZkUtils zkUtils = ZkUtils.apply(zkStr, 30000, 30000, JaasUtils.isZkSecurityEnabled());
  try {
  AdminUtils.createTopic(zkUtils, topic.getTopicName(), topic.getPartition(),
  topic.getReplication(), new Properties(), RackAwareMode.Enforced);
  System.out.printf("create topic:%s success..", topic.getTopicName());
  } catch (Exception e) {
  System.out.printf("create topic error..the root cause:", e.getMessage());
  }
  }

  public static void deleteKafaTopic(String zkStr, KafkaTopicBean topic) {
  ZkUtils zkUtils = ZkUtils.apply(zkStr, 30000, 30000, JaasUtils.isZkSecurityEnabled());
  try {
  AdminUtils.deleteTopic(zkUtils, topic.getTopicName());
  System.out.printf("delete topic:%s success..", topic.getTopicName());
  } catch (Exception e) {
  System.out.printf("create topic error..the root cause:", e.getMessage());
  }
  zkUtils.close();
  }

  public static void updateTopic(String zkStr, KafkaTopicBean topic) {
  ZkUtils zkUtils = ZkUtils.apply(zkStr, 30000, 30000, JaasUtils.isZkSecurityEnabled());
  try {
  Properties props = AdminUtils.fetchEntityConfig(zkUtils, ConfigType.Topic(), topic.getTopicName());
  props.put("min.cleanable.dirty.ratio", "0.3");
  props.remove("max.message.bytes");
  AdminUtils.changeTopicConfig(zkUtils, topic.getTopicName(), props);
  System.out.printf("update topic:%s success..", topic.getTopicName());
  } catch (Exception e) {
  System.out.printf("create topic error..the root cause:", e.getMessage());
  }
  zkUtils.close();
  }

  public static List<String> getAllTopics(String zkStr) {
  ZkUtils zkUtils = ZkUtils.apply(zkStr, 30000, 30000, JaasUtils.isZkSecurityEnabled());
  List<String> topicList = null;
  try {
  List<String> allTopicList = JavaConversions.seqAsJavaList(zkUtils.getAllTopics());
  topicList = allTopicList.stream()
  .filter(topic -> !topic.equalsIgnoreCase("rawMessage") && !topic.equalsIgnoreCase("blmessage"))
  .collect(Collectors.toList());
  } catch (Exception e) {
  System.out.printf("create topic error..the root cause:", e.getMessage());
  }
  return topicList;
  }
  }

测试：
package com.ky.service;

import com.ky.common.KafkaTopicBean;
import com.ky.common.KafkaUtil;

/**

• @Author: xwj

• @Date: 2019/1/7 0007 13:51

• @Version 1.0
  */
  public class KafkaService {

  public static void main(String[] args) {
  String zkStr = "192.168.1.10:2181,192.168.1.11:2181,192.168.1.12:2181";
  String topic = "rawMessage";
  final KafkaTopicBean topicBean = new KafkaTopicBean();
  topicBean.setPartition(1);
  topicBean.setReplication(1);
  topicBean.setTopicName(topic);
  KafkaUtil.createKafaTopic(zkStr, topicBean);
  }
  }

然后去linux上去查看，发现topic已经创建了。
但是删除kafka的topic，只是标记被删除了，并没有真正的删除，要做真正的删除，需要做如下动作。
打开kafka的配置文件，并修改如下属性：

image.png

auto.create.topics.enable 设置为false，不让程序自动创建topic，就是程序再往kafka生产数据的时候，如果发现topic不存在就会自动创建。
delete.topic.enable设置为true，删除topic的同时，会过一段时间删除目录和对应数据文件目录。

方法2：
基于方法1上，代码如下
import org.I0Itec.zkclient.ZkClient;

import java.util.List;

/**

• @Author: xwj

• @Date: 2019/1/7 0007 16:52

• @Version 1.0
  */
  public class ZkService {

  public static void main(String[] args) {
  String zkStr = "192.168.1.10:2181,192.168.1.11:2181,192.168.1.12:2181";
  String path = "/brokers/topics/";
  final List<String> allTopics = KafkaUtil.getAllTopics(zkStr);
  ZkClient zkClient;
  zkClient = new ZkClient(zkStr, 30000, 30000);
  for (String topic : allTopics) {
  try {
  zkClient.deleteRecursive(path + topic);
  System.out.println("delete topic sucess..");
  } catch (Exception e) {
  e.printStackTrace();
  }
  }

  }
  }