package com.elab.example.springbootdemo.kafka;

import com.elab.example.springbootdemo.SpringbootDemoApplication;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.concurrent.Future;

/**
 * @author ： liukx
 * @time ： 2019/8/7 - 13:53
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = {SpringbootDemoApplication.class})
public class KafkaSpringTest {

    @Autowired
    private KafkaProducer<String, String> producer;

    @Test
    public void send() {
        //构造一个消息队列 Kafka 消息
        String topic = "api_log"; //消息所属的 Topic，请在控制台创建后，填写在这里
        String value = "this is the message's value"; //消息的内容
        ProducerRecord<String, String> kafkaMessage = new ProducerRecord<String, String>(topic, value);
        try {
            //发送消息，并获得一个 Future 对象
            Future<RecordMetadata> metadataFuture = producer.send(kafkaMessage);
            //同步获得 Future 对象的结果
            RecordMetadata recordMetadata = metadataFuture.get();
            System.out.println("Produce ok:" + recordMetadata.toString());
        } catch (Exception e) {
            //要考虑重试，参见常见问题: https://help.aliyun.com/document_detail/68168.html
            System.out.println("error occurred");
            e.printStackTrace();
        }
    }


}
