wget https://mirrors.huaweicloud.com/apache/zookeeper/zookeeper-3.5.7/apache-zookeeper-3.5.7-bin.tar.gz

zkServer.sh start
zkServer.sh stop

vim /root/bin/zookeeperstartstop.sh

$ZOOKEEPER_HOME/bin/zkCli.sh -server hadoop102:2181

#KAFKA_HOME
export KAFKA_HOME=/opt/Kafka380/kafka_2.12-3.8.0
export PATH=$PATH:$KAFKA_HOME/bin

vim $KAFKA_HOME/config/server.properties

# 每个服务器上的broker.id要设置得不同
broker.id=2
listeners=PLAINTEXT://192.168.0.45:9092
advertised.listeners=PLAINTEXT://113.45.129.200:9092
log.dirs=/usr/kafaka-logs
zookeeper.connect=192.168.0.45:2181,192.168.11.210:2181,192.168.0.5:2181/kafka

vim /opt/Kafka380/kafka_2.12-3.8.0/bin/kafka-server-start-sasl.sh

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx256M -Xms256M"
fi

$KAFKA_HOME/bin/kafka-topics.sh --create --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --replication-factor 3 --partitions 4 --topic topicDB

bin/kafka-topics.sh --describe --topic topicDB --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092

$KAFKA_HOME/bin/kafka-topics.sh --delete --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --topic topicDB

cd $KAFKA_HOME
bin/kafka-console-consumer.sh --topic mytopic --bootstrap-server 192.168.0.45:9092,192.168.11.210:9092,192.168.0.5:9092

cd $KAFKA_HOME
bin/kafka-console-producer.sh --broker-list 192.168.0.45:9092,192.168.11.210:9092,192.168.0.5:9092 --topic mytopic

try {
    producer.send(record, new Callback() {
        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
            if (e != null) {
                e.printStackTrace();
            } else {
                System.out.println("Topic: " + recordMetadata.topic() + "\n" +
                        "Partition: " + recordMetadata.partition() + "\n" +
                        "Offset: " + recordMetadata.offset());
            }
        }
    });
} catch (Exception e) {
    e.printStackTrace();
} finally {
    producer.close();
}

RecordMetadata recordMetadata = producer.send(record).get();

producer.send(record).get();

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.example.kafkaSerializer.UserQuery;

import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

public class KafkaProducerInterceptor implements ProducerInterceptor<String, UserQuery> {
    private final AtomicLong sendSuccess = new AtomicLong(0);;
    private final AtomicLong sendFailure = new AtomicLong(0);;

    @Override
    public ProducerRecord<String, UserQuery> onSend(ProducerRecord<String, UserQuery> record) {
        System.out.println(record.value());
        System.out.println(record.value().getUserName() + record.value().getUserQuery());
        return new ProducerRecord<>(record.topic(), record.partition(),
                record.timestamp(), record.key(), record.value(), record.headers());
    }

    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
        if (e == null) {
            sendSuccess.incrementAndGet();
        } else {
            sendFailure.incrementAndGet();
        }
    }

    @Override
    public void close() {
        System.out.println("[INFO] 发送成功数：" + sendSuccess.get() + "；发送失败数：" + sendFailure.get());
    }

    @Override
    public void configure(Map<String, ?> map) {}
}

props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, "org.example.kafkaInterceptor.KafkaProducerInterceptor");

props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, "org.example.kafkaInterceptor.KafkaProducerInterceptor1" + "org.example.kafkaInterceptor.KafkaProducerInterceptor2");

{
  "topics": [
    {"topic":  "mytopic"}, {"topic":  "anotherTopic_if_I_have"}
  ],
  "version": 1
}

cd $KAFKA_HOME
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --topics-to-move-json-file topics-to-move.json --broker-list "102,103,104,105" --generate

cd $KAFKA_HOME
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute

cd KAFKA_HOME
bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --verify

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

public class KafkaConsumerAnalysis {
    // public static final String brokerList = "192.168.0.45:9092,192.168.11.210:9092,192.168.0.5:9092";
    public static final String brokerList = "localhost:9092";
    public static final String topic = "topic-demo";
    public static final String groupId = "consumer-group-0 ";
    public static final AtomicBoolean isRunning = new AtomicBoolean(true);

    public static Properties initConfig() {
        Properties props = new Properties();
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumer-client-0");

        return props;
    }

    public static void main(String[] args) {
        Properties props = initConfig();
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        // 订阅主题以及对应的分区
        ArrayList<TopicPartition> topicPartitions = new ArrayList<>();
        topicPartitions.add(new TopicPartition(topic, 0));
        consumer.assign(topicPartitions);
        // 仅订阅某个主题（分区自动分配）
        // consumer.subscribe(Arrays.asList(topic));
        try {
            while (isRunning.get()) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println("topic = " + record.topic() + ", partition = " + record.partition() + ", offset = " + record.offset());
                    System.out.println("key = " + record.key() + ", value = " + record.value());
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }
}

consumer.subscribe(Arrays.asList(this.topic));

consumer.subscribe(Pattern.compile("topic-,*"));

consumer.assign(Arrays.asList(new TopicPartition("topic-demo", 0)));

List<TopicPartition> partitions = new ArrayList<>();
List<PartitionInfo> partitionInfos = consumer.partitionFor(this.topic);
if (partitionInfos != null) {
    for (PartitionInfo tpInfo : partitionInfos) {
        partitions.add(new TopicPartition(tpInfo.topic(), tpInfo.partition()));
    }
}
consumer.assign(partitions);

consumer.unsubscribe();

public class ConsumerRecord<K, V> {
    private final String topic;
    private final int partition;
    private final long offset;
    private final long timestamp;
    private final TimestampType timestampType;
    private final int serializedKeySize;
    private final int serializedValueSize;
    private final Headers headers;
    private final K key;
    private final V value;
    private volatile Long checksum;
    // ...
}

#KE_HOME
export KE_HOME=/opt/KafkaEagle301/efak-web-3.0.1
export PATH=$PATH:$KE_HOME/bin

CREATE DATABASE ke DEFAULT CHARSET utf8mb4 COLLATE utf8mb4_general_ci;

CREATE USER 'ke_user'@'%' IDENTIFIED BY '000000';
GRANT ALL PRIVILEGES ON ke.* TO 'ke_user'@'%';
FLUSH PRIVILEGES;

-- 创建表结构
CREATE TABLE alarm_config (
  id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
  group_name VARCHAR(64) NOT NULL,
  alarm_name VARCHAR(128) NOT NULL,
  alarm_status INT NOT NULL DEFAULT 1,
  alarm_type VARCHAR(32) NOT NULL,
  alarm_level VARCHAR(32) NOT NULL,
  alarm_condition VARCHAR(512),
  alarm_contact_group VARCHAR(512),
  alarm_create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  alarm_modify_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

CREATE TABLE kafka_offsets (
  id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
  cluster_name VARCHAR(64) NOT NULL,
  group_name VARCHAR(128) NOT NULL,
  topic_name VARCHAR(128) NOT NULL,
  partition_id INT NOT NULL,
  log_size BIGINT NOT NULL,
  offsets BIGINT NOT NULL,
  `lag` BIGINT NOT NULL,
  create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE user (
  id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
  username VARCHAR(64) NOT NULL,
  password VARCHAR(64) NOT NULL,
  email VARCHAR(128),
  role VARCHAR(32) NOT NULL DEFAULT 'admin',
  create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

mysql -u ke_user -p ke < $KE_HOME/db/ke.sql

vim $KE_HOME/conf/system-config.properties

# 启动Kafka-Eagle
$KE_HOME/bin/ke.sh start

# 停止Kafka-Eagle
$KE_HOME/bin/ke.sh stop

# 查看Kafka-Eagle状态
$KE_HOME/bin/ke.sh status

# 重启Kafka-Eagle
$KE_HOME/bin/ke.sh restart

ssh -L 8048:192.168.0.45:8048 root@113.45.129.200

# 在Kerberos服务端执行下面这一行命令创建Kafka用户主体并设置好密码
kadmin.local -q "add_principal kafka_user"
# 另外再为kafka_user创建一个keytab（暂且创建在hadoop102上）
kadmin.local -q "ktadd -kt /etc/security/keytabs/kafkauserService.keytab kafka_user@EXAMPLE.COM"
chown kafka_user:hadoop /etc/security/keytabs/kafkauserService.keytab
cp /etc/security/keytabs/kafkauserService.keytab $KAFKA_HOME/config/kerberos

# 在每个节点上执行下面一行命令，并且设置不同节点的kafka_user用户两两之间ssh免密
useradd -g hadoop -m kafka_user

kadmin.local -q "add_principal -randkey kafka/hadoop102@EXAMPLE.COM"
kadmin.local -q "add_principal -randkey kafka/hadoop103@EXAMPLE.COM"
kadmin.local -q "add_principal -randkey kafka/hadoop104@EXAMPLE.COM"

ssh hadoop102
kadmin -p zt/admin@EXAMPLE.COM -w123456 -q "ktadd -kt /etc/security/keytabs/kafkaService.keytab kafka/hadoop102@EXAMPLE.COM"
chown kafka_user:hadoop /etc/security/keytabs/kafkaService.keytab
ssh hadoop103
kadmin -p zt/admin@EXAMPLE.COM -w123456 -q "ktadd -kt /etc/security/keytabs/kafkaService.keytab kafka/hadoop103@EXAMPLE.COM"
chown kafka_user:hadoop /etc/security/keytabs/kafkaService.keytab
ssh hadoop104
kadmin -p zt/admin@EXAMPLE.COM -w123456 -q "ktadd -kt /etc/security/keytabs/kafkaService.keytab kafka/hadoop104@EXAMPLE.COM"
chown kafka_user:hadoop /etc/security/keytabs/kafkaService.keytab

# 启动Kerberos安全模式下的Kafka。(-daemon 后台启动)
cd $KAFKA_HOME
bin/kafka-server-start-sasl.sh -daemon config/server-sasl.properties

# 创建topic
bin/kafka-topics-sasl.sh --create --topic topicDB --bootstrap-server hadoop102:19092,hadoop103:19092,hadoop104:19092  --partitions 4 --replication-factor 3 --config retention.ms=259200000 --command-config config/kerberos/client.properties

# 查看topic
bin/kafka-topics-sasl.sh --bootstrap-server hadoop102:19092,hadoop103:19092,hadoop104:19092 --describe --topic topicDB --command-config config/kerberos/client.properties

# 删除topic
bin/kafka-topics-sasl.sh --delete --topic topicDB --bootstrap-server hadoop102:19092,hadoop103:19092,hadoop104:19092 --command-config config/kerberos/client.properties

# 启动消费者
bin/kafka-console-consumer-sasl.sh --bootstrap-server hadoop102:19092,hadoop103:19092,hadoop104:19092 --topic topicDB --from-beginning --consumer.config config/kerberos/client.properties

# 启动生产者
bin/kafka-console-producer-sasl.sh --bootstrap-server hadoop102:19092,hadoop103:19092,hadoop104:19092 --topic topicDB --producer.config config/kerberos/client.properties

[realms]
    EXAMPLE.COM = {
    kdc=<公网IP>
    admin_server=<公网IP>
    }

props.put("security.protocol", "SASL_PLAINTEXT");
props.put("sasl.mechanism", "GSSAPI");
props.put("sasl.kerberos.service.name", "kafka");
props.put("sasl.jaas.config",
        "com.sun.security.auth.module.Krb5LoginModule required " +
                "useKeyTab=true " +
                "keyTab=\"D:/path/to/kafkauserService.keytab\" " +
                "storeKey=true " +
                "useTicketCache=false " +
                "principal=\"kafka_user@EXAMPLE.COM\";");

System.setProperty("java.security.krb5.conf", "D:\\path\\to\\krb5.ini");