Cloudera 7.13.1 Kafka 브로커(9093 포트, ssl.client.auth=none)에 NiFi 1.28.1 의 ConsumeKafkaRecord 로 붙이려는데 Streams Messaging Manager 에 Consumer Group 이 잡히지 않고, NiFi 로그에 Bootstrap broker ... disconnected 경고만 반복된다. 브로커 인증서는 Cloudera Manager Auto-TLS 가 발급한 것이다.
브로커 프로세스 설정에서 listener 와 보안 프로토콜을 본다.
cat /var/run/cloudera-scm-agent/process/*-kafka-KAFKA_BROKER/kafka.properties | grep -i "listener\|security\|sasl\|ssl"
SSL://0.0.0.0:9093 이면 SSL 단독 연결이 가능하고, SASL_SSL 이면 SASL 인증 정보가 추가로 필요하다. 브로커 인증서 발급자는 다음으로 본다.
openssl s_client -connect <broker>:9093 -showcerts 2>/dev/null | grep -E "subject|issuer"
발급자가 SCM Local CA on ... 이면 CA 번들은 /opt/cloudera/security/cm-auto-global_cacerts.pem 에 있다. 브로커가 쓰는 truststore 파일과 비밀번호를 가져올 필요는 없다. 브로커 인증서를 발급한 CA 만 NiFi 쪽 truststore 에 있으면 된다.
cat > /etc/kafka/conf/client.properties << 'EOT'
security.protocol=SSL
ssl.truststore.type=PEM
ssl.truststore.location=/opt/cloudera/security/cm-auto-global_cacerts.pem
EOT
/opt/cloudera/parcels/CDH/bin/kafka-topics \
--bootstrap-server <broker1>:9093,<broker2>:9093,<broker3>:9093 \
--command-config /etc/kafka/conf/client.properties --list
토픽 목록이 나오면 인증서와 네트워크는 정상이고 남는 것은 NiFi 설정이다.
NiFi 1.28 의 StandardSSLContextService 는 PEM 을 받지 않고 JKS · PKCS12 · BCFKS 만 받는다. PEM 번들에 인증서 블록이 둘 이상이면 블록마다 나눠서 넣는다.
awk '/BEGIN CERTIFICATE/{n++} {print > "/tmp/cert" n ".pem"}' /opt/cloudera/security/cm-auto-global_cacerts.pem
keytool -import -alias kafka-ca-1 -file /tmp/cert1.pem \
-keystore /opt/cloudera/security/kafka-truststore.jks \
-storetype JKS -storepass ${TRUSTSTORE_PASSWORD} -noprompt
keytool -import -alias kafka-ca-2 -file /tmp/cert2.pem \
-keystore /opt/cloudera/security/kafka-truststore.jks \
-storetype JKS -storepass ${TRUSTSTORE_PASSWORD} -noprompt
keytool -list -keystore /opt/cloudera/security/kafka-truststore.jks -storepass ${TRUSTSTORE_PASSWORD}
alias 는 임의로 정해도 된다.
StandardSSLContextService 를 새로 만든다. 기존 서비스를 고쳐도 반영되지 않는 경우가 있어 Disable → Delete 후 재생성이 확실하다.
| 속성 | 값 |
|---|---|
| Truststore Filename | /opt/cloudera/security/kafka-truststore.jks |
| Truststore Password | ${TRUSTSTORE_PASSWORD} |
| Truststore Type | JKS |
| Keystore Filename / Password / Type | 비워 둔다 (단방향 SSL) |
| TLS Protocol | TLS |
ConsumeKafkaRecord 설정은 다음과 같다.
| 속성 | 값 |
|---|---|
| Security Protocol | SSL |
| SASL Mechanism | GSSAPI (필수 선택 항목이라 골라 두고 Kerberos 관련 항목은 전부 비운다) |
| SSL Context Service | 위에서 만든 서비스 |
동적 속성 ssl.endpoint.identification.algorithm |
빈 값 |
Security Protocol 이 SSL 이면 SASL Mechanism 값은 실제로 쓰이지 않는다. PLAIN 을 고르면 사용자·비밀번호가 필수가 되므로 GSSAPI 를 고른다. ssl.endpoint.identification.algorithm 을 비우면 브로커 인증서 CN 과 접속 호스트명이 달라도 통과한다.