kafka-ui:
  image: provectuslabs/kafka-ui
  container_name: kafka-ui
  restart: always
  depends_on:
    - kafka1
  ports:
    - 9091:8080
  volumes:
    - ${DATA_PATH}/kafka_ui/data:/etc/localtime
    - /etc/localtime:/etc/localtime:ro
  environment:
    # 集群名称
    - KAFKA_CLUSTERS_0_NAME=local
    # 集群地址
    - KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=kafka1:9092

kafka1:
  image: 'bitnami/kafka'
  user: root
  ports:
    - 9192:9092
    - 9193:9093
  environment:
    - ALLOW_PLAINTEXT_LISTENER=yes
  volumes:
    - ${DATA_PATH}/kafka1/data:/bitnami/kafka
    - /etc/localtime:/etc/localtime:ro

version: '2'

services:
  kafka-0:
    image: docker.io/bitnami/kafka:testing
    ports:
      - '9092'
      - '9093'
    environment:
      - BRDEBUG=1
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka-0:9093,1@kafka-1:9093,2@kafka-2:9093
      - KAFKA_CFG_BROKER_ID=0
      - KAFKA_KRAFT_CLUSTER_ID=abcdefghijklmnopqrstuv
      - ALLOW_PLAINTEXT_LISTENER=yes
    volumes:
      - kafka_0_data:/bitnami/kafka
  kafka-1:
    image: docker.io/bitnami/kafka:testing
    ports:
      - '9092'
      - '9093'
    environment:
      - BRDEBUG=1
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka-0:9093,1@kafka-1:9093,2@kafka-2:9093
      - KAFKA_CFG_BROKER_ID=1
      - KAFKA_KRAFT_CLUSTER_ID=abcdefghijklmnopqrstuv
      - ALLOW_PLAINTEXT_LISTENER=yes
    volumes:
      - kafka_1_data:/bitnami/kafka
  kafka-2:
    image: docker.io/bitnami/kafka:testing
    ports:
      - '9092'
      - '9093'
    environment:
      - BRDEBUG=1
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka-0:9093,1@kafka-1:9093,2@kafka-2:9093
      - KAFKA_KRAFT_CLUSTER_ID=abcdefghijklmnopqrstuv
      - KAFKA_CFG_BROKER_ID=2
      - ALLOW_PLAINTEXT_LISTENER=yes
    volumes:
      - kafka_2_data:/bitnami/kafka

volumes:
  kafka_0_data:
    driver: local
  kafka_1_data:
    driver: local
  kafka_2_data:
    driver: local

root@kafka-0:/# /opt/bitnami/kafka/bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --topic mytopic --partitions 3 --replication-factor 3
Created topic "mytopic".

root@kafka-0:/# /opt/bitnami/kafka/bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic mytopic
Topic:mytopic   PartitionCount:3        ReplicationFactor:3     Configs:
        Topic: mytopic  Partition: 0    Leader: 2       Replicas: 2,3,1 Isr: 2,3,1
        Topic: mytopic  Partition: 1    Leader: 3       Replicas: 3,1,2 Isr: 3,1,2
        Topic: mytopic  Partition: 2    Leader: 1       Replicas: 1,2,3 Isr: 1,2,3

...
services:
  kafka:
    ...
    volumes:
      - 'kafka_data:/bitnami'
+     - /path/to/server.properties:/bitnami/kafka/config/server.properties

vi /path/to/server.properties

docker-compose restart kafka

docker-compose exec  kafka1  bash
cd /opt/bitnami/kafka

bin/kafka-topics.sh --bootstrap-server localhost:9092  --topic quickstart-events  --create

# 查看topic列表
bin/kafka-topics.sh --bootstrap-server localhost:9092  --list

# 查看topic列表详情
bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092

# 查看消费者组
bin/kafka-consumer-groups.sh --list --bootstrap-server localhost:9092

# 修改分区，扩分区，不能减少分区
bin/kafka-topics.sh --alter --topic quickstart-events --partitions 3 --bootstrap-server localhost:9092
# 修改过期时间，下面两行都可以
bin/kafka-configs.sh --bootstrap-server localhost:9092  --alter --topic quickstart-events --add-config retention.ms=86400000

bin/kafka-configs.sh --bootstrap-server localhost:9092  --alter --entity-name quickstart-events --entity-type topics --add-config retention.ms=86400000

# 删除topic 看看就可以了，别真删除了。
bin/kafka-topics.sh --delete --topic quickstart-events --bootstrap-server localhost:9092

bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic quickstart-events
my first event
my second event

bin/kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092
my first event
my second event

#指定从分区的某个位置开始消费，这里只指定了一个分区，可以多写几行或者遍历对应的所有分区
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic quickstart-events --partition 0 --offset 100

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic quickstart-events --group mygroup

bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group mygroup

bootstrap.servers=localhost:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true
offset.storage.file.filename=/tmp/connect.offsets
offset.flush.interval.ms=10000
plugin.path=libs/connect-file-3.4.0.jar

name=local-file-source
connector.class=FileStreamSource
tasks.max=1
file=test.txt
topic=connect-test

name=local-file-sink
connector.class=FileStreamSink
tasks.max=1
file=test.sink.txt
topics=connect-test

apt update
apt install vim

cd /opt/bitnami/kafka/config
more connect-standalone.properties
vim connect-standalone.properties
# 添加下面内容：plugin.path=libs/connect-file-3.4.0.jar

bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties

> more test.sink.txt
foo
bar

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning
{"schema":{"type":"string","optional":false},"payload":"foo"}
{"schema":{"type":"string","optional":false},"payload":"bar"}

> echo Another line>> test.txt

dependencies {
    implementation group: 'org.apache.kafka', name: 'kafka-clients', version: '3.4.0'
    implementation group: 'org.slf4j', name: 'slf4j-simple', version: '2.0.7'
    testImplementation(platform("org.junit:junit-bom:5.9.1"))
    testImplementation("org.junit.jupiter:junit-jupiter")
}

//缓存区大小
properties.setProperty(ProducerConfig.BUFFER_MEMORY_CONFIG, String.valueOf(33554432));
//批次大小
properties.setProperty(ProducerConfig.BATCH_SIZE_CONFIG,String.valueOf(16384));
//间隔时间
properties.setProperty(ProducerConfig.LINGER_MS_CONFIG, String.valueOf(1));
//压缩
properties.setProperty(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");

//acks，默认是-1
properties.setProperty(ProducerConfig.ACKS_CONFIG, String.valueOf(1));
//重试参数，默认是int最大值
properties.setProperty(ProducerConfig.RETRIES_CONFIG, String.valueOf(3));

public class CustomProducerTransaction {
    public static void main(String[] args){
        //1. 创建kafka生产对象
        Properties properties=CustomProducer.getBaseProperties();
        // 指定一个事务ID，这个ID不重复就可以了，可以自动得到一个雪花ID
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"123456");

        KafkaProducer<String,String> kafkaProducer=new KafkaProducer<>(properties);

        //1.1初始化事务
        kafkaProducer.initTransactions();
        //1.2开启事务
        kafkaProducer.beginTransaction();
        try{
            //2. 发送数据
            for(int i=0;i<5;i++){
                kafkaProducer.send(new ProducerRecord<>("quickstart-events","1CustomProducerTransaction"+i));
            }

            //设置一个错误,
            //int i=2/0;

            //1.3提交事务
            kafkaProducer.commitTransaction();
        }catch (Exception e){
            //1.4回撤事务
            kafkaProducer.abortTransaction();
        }finally {
            //3. 关闭资源
            kafkaProducer.close();
            System.out.println("--------------------");
        }
    }
}

properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,3000);

//指定位置，首先要拿到所有的分区，然后从分区信息中指定offset
Set<TopicPartition> assignment = kafkaConsumer.assignment();
//担心没有拿到分区方案，这里又等待了一下。
while (assignment.size()==0){
    kafkaConsumer.poll(Duration.ofSeconds(1));
    assignment=kafkaConsumer.assignment();
}
for(TopicPartition topicPartition:assignment){
    kafkaConsumer.seek(topicPartition,19470);
}

public class CustomConsumerSeekByTime {
    public static void main(String[] args) throws ParseException {
        //0 配置
        Properties properties = CustomConsumer.getBaseProperties();
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"ConsumerSeekByTime");

        //1 创建一个消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
        //2 订阅一个主题
        Collection<String> topics=new ArrayList<>();
        topics.add("quickstart-events");
        kafkaConsumer.subscribe(topics);

        //指定位置，首先要拿到所有的分区，然后从分区信息中指定offset
        Set<TopicPartition> assignment = kafkaConsumer.assignment();
        //担心没有拿到分区方案，这里又等待了一下。
        while (assignment.size()==0){
            kafkaConsumer.poll(Duration.ofSeconds(1));
            assignment=kafkaConsumer.assignment();
        }

        SimpleDateFormat sdf1=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Date dateTime = sdf1.parse("2023-6-4 15:20:24");
        //把时间抓换成offset
        Map<TopicPartition, Long> timestampsToSearch=new HashMap<>();
        // 填充timestampsToSearch
        for (TopicPartition topicPartition:assignment){
            timestampsToSearch.put(topicPartition,dateTime.getTime());
        }
        Map<TopicPartition, OffsetAndTimestamp> topicPartitionOffsetAndTimestampMap
                =  kafkaConsumer.offsetsForTimes(timestampsToSearch);

        for(TopicPartition topicPartition:assignment){
            OffsetAndTimestamp offsetAndTimestamp = topicPartitionOffsetAndTimestampMap.get(topicPartition);
            kafkaConsumer.seek(topicPartition,offsetAndTimestamp.offset());
        }

        //3 消费数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String,String> consumerRecord:consumerRecords){
                System.out.println(consumerRecord);
            }
        }
    }
}

public class KafkaFlinkProducer {

    public static void main(String[] args) throws Exception {
        //1.获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);
        //2.准备数据源
        //模拟一个集合作为数据源
        ArrayList<String> wordList = new ArrayList<>();
        for(int i=0;i<100;i++){
            wordList.add("test:"+i);
        }
        DataStreamSource<String> stream=env.fromCollection(wordList);
        //创建kafka sink
        KafkaSink<String> kafkaSink=KafkaSink.<String>builder()
                .setBootstrapServers("localhost:9092")
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.builder()
                        .setTopic("quickstart-events")
                        .setValueSerializationSchema(new SimpleStringSchema())
                        .build()
                )
                //.setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
                .build();

        //3.添加数据源
        stream.sinkTo(kafkaSink)
                .name("myKafkaSink")
                .setParallelism(1)
        ;
        env.execute("myJob");
    }
}

public class KafkaFlinkConsumer {
    public static void main(String[] args) throws Exception {
        //1.获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);
        //2.创建消费者
        KafkaSource<String> source=KafkaSource.<String>builder()
                .setBootstrapServers("localhost:9092")
                .setTopics("quickstart-events")
                .setGroupId("my-group")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                //.setStartingOffsets(OffsetsInitializer.earliest())
                .build();

        //3.关联消费者与flink流

        DataStreamSource<String> dataStreamSource=  env.fromSource(source
                , WatermarkStrategy.noWatermarks()
                ,"Kafka Source"
                , TypeInformation.of(String.class)
        );

        //4.执行
//        SingleOutputStreamOperator<String> myCounter = dataStreamSource.process(new ProcessFunction<>() {
//            @Override
//            public void processElement(String s, Context context, Collector<String> collector) {
//                System.out.println(s);
//            }
//        });

        PrintSink<String> printSink = new PrintSink<>(true);
        dataStreamSource.sinkTo(printSink)
                .name("printSink")
                .setParallelism(3)
        ;
        env.execute("Flink DataSource");
    }
}

spring:
  kafka:
    bootstrap-servers: localhost:9092

@Service
public class KafkaProducer {
    @Autowired
    private KafkaTemplate kafkaTemplate;
    public String send(String value){
        assert value!=null;
        CompletableFuture<SendResult<String,String>> completableFuture=
                kafkaTemplate.send("quickstart-events",value);
        completableFuture.thenAccept(result->{
            System.out.println("send ok:"+value);

        });
        completableFuture.exceptionally(e->{
            System.out.println(e.toString());
            return null;
        });
        return "ok";
    }
}

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: 'springGroup'

@Configuration
public class KafkaConsumer {
    @KafkaListener(topics = "quickstart-events")
    public void consumer(String value){
        System.out.println("收到kafka:"+value);
    }
}

bin/kafka-topics.sh --bootstrap-server localhost:9002 --create --topic test --rtitions 3 --replication-factor 3

bin/kafka-producer-perf-test.sh --topic test --record-size 1024 --num-recorder 1000000 --throughput 30000 --producer-props bootstrap.servers=localhost:9092 batch.size=16384 linger.ms=0

bin/kafka-consumer-perf-test.sh --bootstrap-server localhost:9092 --topic test --messages 10000000 ---consumer.config config/consumer.properties

max.poll.records=500