Databricks にて Confluent からデータを取得する際の Failed to create new KafkaAdminClient エラーへの対応方法



## 概要

Databricks 上で Confluent（Kafka）からデータを取得しようとすると、まれに以下のようなエラーが発生することがあります。本記事では、よくあるエラーパターンや、その再現方法・対応策を紹介します。

```text
kafkashaded.org.apache.kafka.common.KafkaException: Failed to create new KafkaAdminClient
```



## エラーパターンと対応方法

### 1. 認証情報が適切でない場合

#### エラーの再現方法

以下のように、誤った Confluent の認証情報を設定して Spark のストリーミング読み込みを行うと、KafkaAdminClient の作成に失敗し、上記エラーが発生するケースがあります。

```python
# Confluent に対する認証情報をセット（誤った例）
bootstrap_servers = "pkc-56d.eastus.azure.confluent.cloud:9092"
sasl_username = "AAAA"
sasl_password = "AAAAA"

kafka_options = {
    "kafka.bootstrap.servers": bootstrap_servers,
    "kafka.security.protocol": "SASL_SSL",
    "kafka.sasl.jaas.config": f'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="{sasl_username}" password="{sasl_password}";',
    "kafka.sasl.mechanism": "PLAIN",
    "subscribe": topic_name,
    "startingOffsets": "earliest",
}

df = (
    spark.readStream.format("kafka")
    .options(**kafka_options)
    .load()
)
df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3346e968-7cfc-d66e-6475-f36ad2dc4a7d.png)

#### 対応方法

Confluent から正しい認証情報（`bootstrap_servers` や `username`, `password` など）を取得して、改めて以下のように設定し直します。正しい認証情報を設定することで、エラーが解消され、Kafka からメッセージを正常に取得できます。

```python
# Confluent に対する認証情報をセット（正しい例）
bootstrap_servers = "pkc-921jm.us-east-2.aws.confluent.cloud:9092"
sasl_username = "322PJOFHAYLZYGMQ"
sasl_password = "8kJNn1c4yuRu41a8/gdcuHGcYEFBkQTOlgTk1luCbTJUBBV5XGNwuUmDou/KE3jM"

kafka_options = {
    "kafka.bootstrap.servers": bootstrap_servers,
    "kafka.security.protocol": "SASL_SSL",
    "kafka.sasl.jaas.config": f'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="{sasl_username}" password="{sasl_password}";',
    "kafka.sasl.mechanism": "PLAIN",
    "subscribe": topic_name,
    "startingOffsets": "earliest",
}

df = (
    spark.readStream.format("kafka")
    .options(**kafka_options)
    .load()
)
df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d2f1ab12-60ab-b131-06db-2eedcd4aefe1.png)


### 2. `kafka.sasl.jaas.config` の末尾が `;` で終わっていない場合

#### エラーの再現方法

`sasl.jaas.config` が正しく設定されていても、以下のように末尾にセミコロン `;` を付け忘れると、KafkaAdminClient の作成に失敗してしまいます。

```python
# Confluent に対する認証情報をセット
bootstrap_servers = "pkc-921jm.us-east-2.aws.confluent.cloud:9092"
sasl_username = "322PJOFHAYLZYGMQ"
sasl_password = "8kJNn1c4yuRu41a8/gdcuHGcYEFBkQTOlgTk1luCbTJUBBV5XGNwuUmDou/KE3jM"

kafka_options = {
    "kafka.bootstrap.servers": bootstrap_servers,
    "kafka.security.protocol": "SASL_SSL",
    "kafka.sasl.jaas.config": f'org.apache.kafka.common.security.plain.PlainLoginModule required username="{sasl_username}" password="{sasl_password}"',
    "kafka.sasl.mechanism": "PLAIN",
    "subscribe": topic_name,
    "startingOffsets": "earliest",
}

df = (
    spark.readStream.format("kafka")
    .options(**kafka_options)
    .load()
)
df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/22da9bba-079f-2077-79dc-f1c343b54555.png)

#### 対応方法

セミコロン `;` が抜けていることが原因です。次のように修正しましょう。

```diff
- "kafka.sasl.jaas.config": f'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="{sasl_username}" password="{sasl_password}"',
+ "kafka.sasl.jaas.config": f'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="{sasl_username}" password="{sasl_password}";',
```

修正後のサンプルコードは以下のとおりです。

```python
kafka_options = {
    "kafka.bootstrap.servers": bootstrap_servers,
    "kafka.security.protocol": "SASL_SSL",
    "kafka.sasl.jaas.config": f'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="{sasl_username}" password="{sasl_password}";',
    "kafka.sasl.mechanism": "PLAIN",
    "subscribe": topic_name,
    "startingOffsets": "earliest",
}

df = (
    spark.readStream.format("kafka")
    .options(**kafka_options)
    .load()
)
df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/55d9a932-ee98-a888-da8e-74606c1e9a2c.png)



### 3. `kafka.sasl.jaas.config` のオプションにて `kafkashaded.` が抜けている場合

#### エラーの再現方法

次のように、Kafka のライブラリによる影響を受けている Databricks 上では、クラス名に `kafkashaded.` が必要です。`org.apache.kafka.common.security.plain.PlainLoginModule` のままで設定すると、同様にエラーが発生します。

```python
# Confluent に対する認証情報をセット（kafkashaded. が抜けた例）
bootstrap_servers = "pkc-921jm.us-east-2.aws.confluent.cloud:9092"
sasl_username = "322PJOFHAYLZYGMQ"
sasl_password = "8kJNn1c4yuRu41a8/gdcuHGcYEFBkQTOlgTk1luCbTJUBBV5XGNwuUmDou/KE3jM"

kafka_options = {
    "kafka.bootstrap.servers": bootstrap_servers,
    "kafka.security.protocol": "SASL_SSL",
    "kafka.sasl.jaas.config": f'org.apache.kafka.common.security.plain.PlainLoginModule required username="{sasl_username}" password="{sasl_password}";',
    "kafka.sasl.mechanism": "PLAIN",
    "subscribe": topic_name,
    "startingOffsets": "earliest",
}

df = (
    spark.readStream.format("kafka")
    .options(**kafka_options)
    .load()
)
df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/de183409-71d7-a6ba-36b0-340fa7f73284.png)

#### 対応方法

次のように `kafkashaded.` 付きのクラス名に修正すると、エラーを解消できます。

```diff
- "kafka.sasl.jaas.config": f'org.apache.kafka.common.security.plain.PlainLoginModule required username="{sasl_username}" password="{sasl_password}";',
+ "kafka.sasl.jaas.config": f'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="{sasl_username}" password="{sasl_password}";',
```

修正後のサンプルコードは以下のとおりです。

```python
kafka_options = {
    "kafka.bootstrap.servers": bootstrap_servers,
    "kafka.security.protocol": "SASL_SSL",
    "kafka.sasl.jaas.config": f'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="{sasl_username}" password="{sasl_password}";',
    "kafka.sasl.mechanism": "PLAIN",
    "subscribe": topic_name,
    "startingOffsets": "earliest",
}

df = (
    spark.readStream.format("kafka")
    .options(**kafka_options)
    .load()
)
df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ecbee7d3-064b-3c54-cf50-ad54aa503160.png)