apache kafka

kafka install on the vagrant

vagrant up

vagrant ssh

https://kafka.apache.org/quickstart

https://www.apache.org/dyn/closer.cgi?path=/kafka/1.1.0/kafka_2.11-1.1.0.tgz

---

1. download

vagrant@single:~$ wget http://apache.mirror.cdnetworks.com/kafka/1.1.0/kafka_2.11-1.1.0.tgz .

--2018-07-02 05:15:02--  http://apache.mirror.cdnetworks.com/kafka/1.1.0/kafka_2.11-1.1.0.tgz

Resolving apache.mirror.cdnetworks.com (apache.mirror.cdnetworks.com)... 14.0.101.165

Connecting to apache.mirror.cdnetworks.com (apache.mirror.cdnetworks.com)|14.0.101.165|:80... connected.

HTTP request sent, awaiting response... 200 OK

Length: 56969154 (54M) [application/x-gzip]

Saving to: `kafka_2.11-1.1.0.tgz'

100%[=====================================================================================================================>] 56,969,154  18.8M/s   in 2.9s

2018-07-02 05:15:05 (18.8 MB/s) - `kafka_2.11-1.1.0.tgz' saved [56969154/56969154]

--2018-07-02 05:15:05--  http://./

Resolving . (.)... failed: Name or service not known.

wget: unable to resolve host address `.'

FINISHED --2018-07-02 05:15:05--

Total wall clock time: 3.1s

Downloaded: 1 files, 54M in 2.9s (18.8 MB/s)

vagrant@single:~$ ls -lrt

total 99116

drwxr-xr-x 14 vagrant vagrant     4096 Jul 22  2013 hadoop-1.2.1

-rw-rw-r--  1 vagrant vagrant 38096663 Jun 20  2017 hadoop-1.2.1-bin.tar.gz

-rw-rw-r--  1 vagrant vagrant 56969154 Mar 28 12:05 kafka_2.11-1.1.0.tgz

-rw-rw-r--  1 vagrant vagrant  6419436 Mar 28 12:05 kafka-1.1.0-src.tgz

vagrant@single:~$ tar -xzf kafka_2.11-1.1.0.tgz

vagrant@single:~$ cd kafka_2.11-1.1.0/

2. start zookeeper

vagrant@single:~/kafka_2.11-1.1.0$ ll

total 60

drwxr-xr-x  6 vagrant vagrant  4096 Mar 23 22:54 ./

drwxr-xr-x 12 vagrant vagrant  4096 Jul  2 05:15 ../

drwxr-xr-x  3 vagrant vagrant  4096 Mar 23 22:54 bin/

drwxr-xr-x  2 vagrant vagrant  4096 Mar 23 22:54 config/

drwxr-xr-x  2 vagrant vagrant  4096 Jul  2 05:15 libs/

-rw-r--r--  1 vagrant vagrant 28824 Mar 23 22:51 LICENSE

-rw-r--r--  1 vagrant vagrant   336 Mar 23 22:51 NOTICE

drwxr-xr-x  2 vagrant vagrant  4096 Mar 23 22:54 site-docs/

vagrant@single:~/kafka_2.11-1.1.0$ ./bin/zookeeper-server-start.sh config/zookeeper.properties

---

3. start server0-2

server_0

vagrant@single:~/kafka_2.11-1.1.0$ ll

total 64

drwxr-xr-x  7 vagrant vagrant  4096 Jul  2 05:15 ./

drwxr-xr-x 12 vagrant vagrant  4096 Jul  2 05:15 ../

drwxr-xr-x  3 vagrant vagrant  4096 Mar 23 22:54 bin/

drwxr-xr-x  2 vagrant vagrant  4096 Mar 23 22:54 config/

drwxr-xr-x  2 vagrant vagrant  4096 Jul  2 05:15 libs/

-rw-r--r--  1 vagrant vagrant 28824 Mar 23 22:51 LICENSE

drwxrwxr-x  2 vagrant vagrant  4096 Jul  2 05:15 logs/

-rw-r--r--  1 vagrant vagrant   336 Mar 23 22:51 NOTICE

drwxr-xr-x  2 vagrant vagrant  4096 Mar 23 22:54 site-docs/

vagrant@single:~/kafka_2.11-1.1.0$ ./bin/kafka-server-start.sh config/server.properties

server_1

config/server-1.properties:

    broker.id=1

    listeners=PLAINTEXT://192.168.127.100:9093

    log.dir=/tmp/kafka-logs-1

 

server_2

config/server-2.properties:

    broker.id=2

    listeners=PLAINTEXT://192.168.127.100:9094

    log.dir=/tmp/kafka-logs-2




설정을 copy후, 리스너 포트를 다르게 하고, 띄운다.

리모트에서 리슨할수있도록 서버호스트도 적어준다.. 

0.0.0.0 로 해서 올렸더니 아래같은 에러.. 

requirement failed: advertised.listeners cannot use the non routable meta-address 0.0.0.0. Use a routable IP address.

---

vagrant@single:~/kafka_2.11-1.1.0$ ./bin/kafka-server-start.sh config/server_1.properties & ; ./bin/kafka-server-start.sh config/server_2.properties &

9092-4 다른 포트로 떴는지 확인

vagrant@single:~$ netstat -nlp | grep 909 (Not all processes could be identified, non-owned process info will not be shown, you would have to be root to see it all.) tcp 0 0 127.0.0.1:29202 0.0.0.0:* LISTEN 909/python tcp6 0 0 :::9092 :::* LISTEN 6174/java tcp6 0 0 :::9093 :::* LISTEN 7229/java tcp6 0 0 :::9094 :::* LISTEN 8110/java

---

4. 토픽명 my-replic topic생성

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replic

5. kafka-producer 스크립트 실행 & kafka-consume 2개를 실행

*producer 실행

vagrant@single:~/kafka_2.11-1.1.0$ ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replic

>blabla

>this is a msg. ------- 컨슘2까지 올리고 실행해보면

*consumer 1,2 실행

vagrant@single:~/kafka_2.11-1.1.0$ bin/kafka-console-consumer.sh --bootstrap-server 192.168.127.100:9092 --from-beginning --topic my-replic

blabla

다른 콘솔에서.. 또 컨슘!

vagrant@single:~/kafka_2.11-1.1.0$ bin/kafka-console-consumer.sh --bootstrap-server 192.168.127.100:9092 --from-beginning --topic my-replic

blabla

this is a msg.

컨슘2도 메시지를 처리 한다.

운영 상태에서.. 신규 컨슈머를 추가 투입하면, 기존 메시지를 한꺼번에 subscribe 하기 떄문에, 새로 유입되는 메시지부터 받을수 있도록 변경해주어야 한다.

---

6. consumer.properties autoOffsetReset = latest

autoOffsetReset:latest

# What to do when there is no initial offset in Kafka or if the current # offset does not exist any more on the server: latest, earliest, none #auto.offset.reset=

설정을 consumer.properties 아래처럼 변경한다.

vagrant@single:~/kafka_2.11-1.1.0/config$ cat consumer.properties | grep -v "^#" bootstrap.servers=localhost:9092 group.id=test-consumer-group auto.offset.reset=latest

kafka-console-consumer.sh  옵션으로는 --from-begging 아니라 --new-consumer 로..

vagrant@single:~/kafka_2.11-1.1.0$ bin/kafka-console-consumer.sh --bootstrap-server 192.168.127.100:9092 --new-consumer --topic pvout The --new-consumer option is deprecated and will be removed in a future major release.The new consumer is used by default if the --bootstrap-server option is provided. newnew?? blabla

group으로 관리를 하게 되는경우? 컨슘은

vagrant@single:~/kafka_2.11-1.1.0$ ./bin/kafka-console-consumer.sh --bootstrap-server 192.168.127.100:9092 --from-beginning --topic jsa-springboot-topic --group jsa-group hello hello1 hello2

---

7. topic --describe

vagrant@single:~/kafka_2.11-1.1.0$ ./bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic pvout Topic:pvout PartitionCount:1 ReplicationFactor:1 Configs: Topic: pvout Partition: 0 Leader: 0 Replicas: 0 Isr: 0

8. topic list는?

vagrant@single:~/kafka_2.11-1.1.0$ ./bin/kafka-topics.sh --list --zookeeper localhost:2181 __consumer_offsets jsa-springboot-topic my-replic pvout