kafka set credentials

https://developer.ibm.com/technologies/messaging/tutorials/kafka-authn-authz/

를 보고 정리.. 잘 작동한다.

broker side.

1. jass파일을 만든다. 

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin"
    user_alice="alice"
    user_bob="bob"
    user_charlie="charlie";
};

이처럼 만들면 사용자가 사용할수 있는 alice, bob, charlie가 추가된다. jass.conf라고 이름짓고 ~/kafka/config안에 저장해놓는다. 

username="admin", password="admin"의 admin/admin은 카프카 브로커끼리의 통신에 사용된다고 하고,

user_alice="alice"의 alice/alice는 유저가 접속할 계정 이름이다. user_{$접속할 아이디}="{$비밀번호}" 포맷이다.

2. kafka 서버 실행 파일인 kafka-server-start.sh을 실행할때 위의 정보를 물고 들어가도록 환경 변수를 설정하자.

https://stackoverflow.com/questions/40890821/how-to-add-jvm-parameters-to-kafka 를 보면 $KAFKA_OPTS를 설정하면 된다. 

쉘파일 내부에 적절하게 export KAFKA_OPTS="-Djava.security.auth.login.config=경로/kafka/config/jass.conf" 처럼 추가해놓자. 그럼 실행시 위의 인증파일을 잘 물고 시행한다.

또는 맨 위의 링크에 나온것처럼 맨 아래의 exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@" 부분에 옵션을 직접 추가해서 exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=$base_dir/../config/jaas-kafka-server.conf kafka.Kafka "$@"

처럼 변경해도 된다.

3. 인증을 사용한다고 프로토콜을 설정해야 한다.

'경로/config/server.properties'를 열고 아래 프로토콜을 설정해준다.

listeners=SASL_PLAINTEXT://:9092

security.inter.broker.protocol=SASL_PLAINTEXT

sasl.mechanism.inter.broker.protocol=PLAIN

sasl.enabled.mechanisms=PLAIN

브로커쪽은 끝

client side

클라이언트에서 인증 없이 접속하면 다음과 같은 에러가 나온다. 

(id: -1 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)

1. 클라이언트 쪽에서도 jaas파일을 만든다.(서버와 약간 다름)

KafkaClient {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="alice"
    password="alice";
};

보면 알겠지만 alice/alice로 접근하겠다는 뜻이다. 이걸 '경로/kafka/config/my-credential.conf' 처럼 저장해놓자.

2. 클라이언트쪽에서도 프로토콜 설정을 해줘야 한다. '경로/kafka/config/my-sasl.properties'를 만들고

security.protocol=SASL_PLAINTEXT

sasl.mechanism=PLAIN

처럼 입력하고 저장해놓는다. 

3. 브로커쪽에서 했었던 KAFKA_OPTS를 추가해주는 작업이다. 

만약 ~/kafka/bin/kafka-consumer-groups.sh을 사용하고 싶으면 그 파일을 열고 

export KAFKA_OPTS="-Djava.security.auth.login.config=$(dirname $0)/../config/my-credential.conf" 를 맨 마지막 exec위에 추가해준다.

4. my-sasl.properties를 만들어놨었으니 실행시 옵션으로 줘서 사용하도록 설정하면 된다.

./kafka-consumer-groups.sh -bootstrap-server 서버1:9092,서버2:9092,서버3:9092 -group {그룹명} --describe -command-config ../config/my-sasl.properties 이런식으로 properties를 줘서 실행시켰을시 결과가 잘 나온다.

% 참고로 kafka-consumer-groups.sh에게 config를 줄때 command-config 로 줬는데 producer라던가 consumer라던가 다른 쉘파일은 옵션명이 또 다르다.

bin/sasl-kafka-console-producer-alice.sh --broker-list localhost:9092 --topic test --producer.config config/sasl-producer-alice.properties

bin/sasl-kafka-console-consumer-bob.sh --bootstrap-server localhost:9092 --topic test --group bob-group --consumer.config config/sasl-consumer-bob.properties

bin/sasl-kafka-consumer-groups-charlie.sh --bootstrap-server localhost:9092 --describe --group bob-group --command-config config/sasl-consumergroup-charlie.properties

추가참고 : https://docs.cloudera.com/documentation/kafka/latest/topics/kafka_command_line.html