Confluent の公式トレーニングコースである Apache Kafka® for Python Developers を Databricks で実行してみた

### 概要

Confluet の公式トレーニングコースである`Apache Kafka® for Python Developers`におけるハンズオンを Databricks で実行する手順を紹介します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c0f1c570-2182-3f4f-036f-706ee2482f83.png)

 引用元：[Introduction to Python for Kafka](https://developer.confluent.io/courses/kafka-python/intro/)

対象のハンズオンは下記です。

1. [Hands On: Setup the Exercise Environment for Confluent Cloud and Python](https://developer.confluent.io/courses/kafka-python/exercise-environment-setup-hands-on/)
2. [Hands On: Use the Python Producer Class](https://developer.confluent.io/courses/kafka-python/producer-class-hands-on/#add-required-imports)
3. [Hands on: Use the Python Consumer Class](https://developer.confluent.io/courses/kafka-python/consumer-class-hands-on/)
4. [Hands On: Use the Python Producer Class with Schemas](https://developer.confluent.io/courses/kafka-python/producer-class-with-schemas-hands-on/)
5. [Hands On: Use the Python Consumer Class with Schemas](https://developer.confluent.io/courses/kafka-python/consumer-class-with-schemas-hands-on/)
6. [Hands On: Use the Python AdminClient Class](https://developer.confluent.io/courses/kafka-python/adminclient-class-hands-on/)

## 手順

### Hands On: Setup the Exercise Environment for Confluent Cloud and Python

#### Sign up for Confluent Cloud

Confluent の無償トライアルに申し込みます。Google アカウントでもうしこんだところ、すぐにアカウントが払い出されました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/bcbf95ca-6691-83b1-abcb-655cc723bde0.png)

引用元：[無料版の Confluent をお試しください：あらゆるクラウドでお使いいただけるマネージド型 Kafka です | JP](https://www.confluent.io/ja-jp/get-started/)

#### Login to the Confluent Cloud Console

Confluent 上で `learn-kafka-python` という environment を作成します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2e9b88c5-ed4e-2dd9-1f6c-2dbccdf5576b.png)

#### Enable Streams Governance

environment 作成時に、`Stream Governance package`の設定を`Advanced`に設定する必要があるようです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2e9b88c5-ed4e-2dd9-1f6c-2dbccdf5576b.png)

`Essentials`で作成してしまった場合には、 environment のページにて右側にある`Stream Governance package`の`Upgrade now`にて`advanced governance controls`を有効にします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/729c451b-4487-2380-ebec-c839473f2350.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9650723d-4eab-2c39-f6b7-40299ed8bff5.png)

#### Create a Cluster in the `learn-kafka-python` Environment

`kafka-python`という名称の cluster を作成します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c35360aa-3f3b-0250-b4ec-ec3ba7949405.png)

#### Create and Download Python Client Configuration Properties

Confluent のサイトでは client から API を取得する手順となっておりますが、 API Keys を取得するための手順であるため`API Keys`からキーの生成をしました。`Download and continue`を選択すると`Boostrap server`の値も取得できるため便利です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fdb08960-75c4-6fa1-114b-a71582bc018a.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/93e68dee-9436-be2f-a3c8-f6d498d3e486.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3bc0c2f3-173e-b209-f147-1c81a5b65886.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/eae47b0c-1b42-f894-f492-37edbe39951b.png)


#### Create Python Dictionary

Databricks 上にノートブックと同じディレクトリ上に`config.py`というファイル名で下記のコードを記述します。認証情報については前の手順で取得した値をセットします。

```python
config = {
    "bootstrap.servers": "<bootstrap-server-endpoint>",
    "security.protocol": "SASL_SSL",
    "sasl.mechanisms": "PLAIN",
    "sasl.username": "<CLUSTER_API_KEY>",
    "sasl.password": "<CLUSTER_API_SECRET>",
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e74f07f8-32e4-e1a2-f175-992a123b91f3.png)

Databricks のノートブック上で下記のコードを実行して想定通りの値が表示されることを確認します。

```python
dbutils.library.restartPython()
from config import config
print(config)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/11f43c0d-76ef-b89e-b1a2-e2dbbe099d59.png)

#### Install Python 3.X and Related Prerequisites on Local Machine

Databricks 上でシングルノードのクラスターを作成します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/79267d32-a5e6-1124-7398-2faff03b7fe8.png)

#### Create a New Virtual Environment with `virtualenv`

本手順は省略します。

#### Install Confluent Kafka

Databricks ノートブック上で下記のコードを実行します。

```python
%pip install confluent-kafka -q
dbutils.library.restartPython()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f317f251-bb29-321d-0282-d6285674fe1a.png)

### Hands On: Use the Python Producer Class

#### Create Topics

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/16147776-31d8-5a43-d128-1b72dd93a2ab.png)

#### Project Setup

Databricks 利用時には不要な手順であるため省略。

#### Add Required Imports

Databricks のノートブック上で下記コードを実行します。

```python
from confluent_kafka import Producer
from config import config
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6c8c838f-26e0-00c3-efee-9deced404ebe.png)


#### Create callback function

Databricks のノートブック上で下記コードを実行します。

```python
def callback(err, event):
    if err:
        print(f'Produce to topic {event.topic()} failed for event: {event.key()}')
    else:
        val = event.value().decode('utf8')
        print(f'{val} sent to partition {event.partition()}.')
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/64a49960-8549-4936-1c87-87cd04fe2f5b.png)

#### Create function to produce to hello_topic

Databricks のノートブック上で下記コードを実行します。

```python
def say_hello(producer, key):
    value = f'Hello {key}!'
    producer.produce('hello_topic', value, key, on_delivery=callback)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9dffd3f7-fbb1-568b-e104-270d07ed09a8.png)

#### Add Main Block

Databricks のノートブック上で下記コードを記述します。

```python
if __name__ == '__main__':
    producer = Producer(config)
    keys = ['Amy', 'Brenda', 'Cindy', 'Derrick', 'Elaine', 'Fred']
    [say_hello(producer, key) for key in keys]
    producer.flush()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/bb419106-5588-6592-2abe-7e90d41ce848.png)

#### Run the Program

Databricks 上で前の手順を実行します。実行後に Confluent 上でデータが書きこまれたことを確認します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e6da887a-4070-372e-2ca5-ae11d689ceb8.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0a3cf485-9aeb-e95d-9d7c-7bc2e445ed2a.png)

### Hands on: Use the Python Consumer Class

#### Use Consumer to Read Events from Kafka

特に手順なし。

#### Topics

特に手順なし。

#### Project Setup

Databricks 利用時には不要な手順であるため省略。

#### Add Required Imports


```python
from confluent_kafka import Consumer, KafkaException
from config import config
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/227c3540-aa48-63c6-0fe1-f81451fdd093.png)


#### Create function to update configuration


```python
def set_consumer_configs():
    config['group.id'] = 'hello_group'
    config['auto.offset.reset'] = 'earliest'
    config['enable.auto.commit'] = False
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/23681afc-e753-51a0-5905-59d7ecd8b185.png)

#### Create callback function for partition assignment


```python
def assignment_callback(consumer, partitions):
    for p in partitions:
        print(f'Assigned to {p.topic}, partition {p.partition}')
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0b00dc47-b0f6-7f0d-6d95-bedf028d9d69.png)

#### Add Main Block


```python
if __name__ == '__main__':
    set_consumer_configs()
    consumer = Consumer(config)
    consumer.subscribe(['hello_topic'], on_assign=assignment_callback)
    try:
        while True:
            event = consumer.poll(1.0)
            if event is None:
                continue
            if event.error():
                raise KafkaException(event.error())
            else:
                val = event.value().decode('utf8')
                partition = event.partition()
                print(f'Received: {val} from partition {partition}    ')
                # consumer.commit(event)
    except KeyboardInterrupt:
        print('Canceled by user.')
    finally:
        consumer.close()

```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7124e076-deda-a70c-cb17-9e5590b83eaf.png)


#### Run the Consumer

前の手順で記述したコードを実行します。処理が継続実行されるため、セルの実行をとめることを忘れないようでしてください。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/24202b30-28a2-f28f-2487-696b0e93848c.png)

#### Produce New Events

省略します。本手順を実施したい場合には、`Hands On: Use the Python Producer Class`と`Hands on: Use the Python Consumer Class`を異なるノートブックで実装してください。下記のコードはノートブックスコープでのライブラリインストールであるため、２つのノートブックの先頭に記述するようにしてください。

```python
%pip install confluent-kafka -q
dbutils.library.restartPython()
```

#### Observe Consumer Group Rebalance Behavior

省略します。本手順を実施したい場合には、前の手順に記述した内容を実施してください。

### Hands On: Use the Python Producer Class with Schemas

#### Prerequisites

Confluent 上で`temp_readings`という topic を作成します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ab45d7b2-1e25-b486-62c7-805cb24a7473.png)

Confluent にて environment の画面に右下にある`Stream Governance API`の`Add key`を選択して、 Schema Registry API の認証情報を取得します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e92c6dda-29b7-d620-3d88-00614600e498.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3ce7308a-d4df-d4bb-fed4-5f27d6cd1655.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0d0b4a48-cba7-a940-7345-d7bc1d4ec0e3.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3d9f83fd-0b4d-71e8-5dda-def39b3cf96d.png)

Databricks のノートブック上で下記のコードを実行します。

```python
%pip install jsonschema -q
%pip install requests -q
dbutils.library.restartPython()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6885ccd6-71e9-b52c-0532-7d0bc042e484.png)

前の手順で作成した`config.py`というファイルに、 Schema Registry API に対する認証情報を追記します。

```python
sr_config = {
    'url': '<schema.registry.url>',
    'basic.auth.user.info':'<SR_API_KEY>:<SR_API_SECRET>'
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/663d9850-3b71-8ee6-f4fe-dba3433b2bd7.png)


#### Project Setup

Databricks 利用時には不要な手順であるため省略。

#### Add Required Imports

```python
from confluent_kafka import Producer
from confluent_kafka.serialization import SerializationContext, MessageField
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.json_schema import JSONSerializer
from config import config, sr_config
import time
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/34bbc467-1f6f-3b9d-2f0b-8bb67b00fda2.png)

#### Define our class and schema

Databricks のノートブック上で下記のコードを実行します。

```python
class Temperature(object):
    def __init__(self, city, reading, unit, timestamp):
        self.city = city
        self.reading = reading
        self.unit = unit
        self.timestamp = timestamp

schema_str = """{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "title": "Temperature",
  "description": "Temperature sensor reading",
  "type": "object",
  "properties": {
    "city": {
      "description": "City name",
      "type": "string"
    },
    "reading": {
      "description": "Current temperature reading",
      "type": "number"
    },
    "unit": {
      "description": "Temperature unit (C/F)",
      "type": "string"
    },
    "timestamp": {
      "description": "Time of reading in ms since epoch",
      "type": "number"
    }
  }
}"""

def temp_to_dict(temp, ctx):
    return {"city":temp.city, 
            "reading":temp.reading,
            "unit":temp.unit, 
            "timestamp":temp.timestamp}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b7e0c1c2-e0e0-d437-5fc3-4f686d4549d6.png)


#### Create some test data

Databricks のノートブック上で下記のコードを実行します。

```python
data = [Temperature('London', 12, 'C', round(time.time()*1000)),
        Temperature('Chicago', 63, 'F', round(time.time()*1000)),
        Temperature('Berlin', 14, 'C', round(time.time()*1000)),
        Temperature('Madrid', 18, 'C', round(time.time()*1000)),
        Temperature('Phoenix', 78, 'F', round(time.time()*1000))]
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/48399c7a-b9be-c6c0-7db8-390847fc22df.png)

#### Create a producer callback function

Databricks のノートブック上で下記のコードを実行します。

```python
def delivery_report(err, event):
    if err is not None:
        print(f'Delivery failed on reading for {event.key().decode("utf8")}: {err}')
    else:
        print(f'Temp reading for {event.key().decode("utf8")} produced to {event.topic()}')
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b78d728b-b849-7272-47b6-9dd02cc37900.png)

#### Add Main Block

Databricks のノートブック上で下記のコードを記述します。

```python
if __name__ == '__main__':
    topic = 'temp_readings'
    schema_registry_client = SchemaRegistryClient(sr_config)

    json_serializer = JSONSerializer(schema_str,
                                     schema_registry_client,
                                     temp_to_dict)

    producer = Producer(config)
    for temp in data:
        producer.produce(topic=topic, key=str(temp.city),
                         value=json_serializer(temp, 
                         SerializationContext(topic, MessageField.VALUE)),
                         on_delivery=delivery_report)

    producer.flush()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ffea5936-3319-1089-12c6-bbd3672d9cfc.png)

#### Execute the program

前の手順のコードを実行後、 Confluent 上でデータが書きこまれたことを確認します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ae1c0389-53d4-c09d-85f0-268b7b7f2493.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/eaa4637a-a999-cbf7-60c8-3e55146a17ad.png)

### Hands On: Use the Python Consumer Class with Schemas

#### Project Setup

Databricks 利用時には不要な手順であるため省略。

#### Add Required Imports

Databricks のノートブック上で下記のコードを実行します。

```python
from confluent_kafka import Consumer
from confluent_kafka.serialization import SerializationContext, MessageField
from confluent_kafka.schema_registry.json_schema import JSONDeserializer
from config import config
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a85882c6-43bd-0270-1fba-542d401e1d28.png)

#### Add consumer specific configuration properties

Databricks のノートブック上で下記のコードを実行します。

```python
def set_consumer_configs():
    config['group.id'] = 'temp_group'
    config['auto.offset.reset'] = 'earliest'
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/036f11de-2cd5-d44f-3ef0-f381cc6c29f5.png)


#### Define our class and schema

Databricks のノートブック上で下記のコードを実行します。

```python
class Temperature(object):
    def __init__(self, city, reading, unit, timestamp):
        self.city = city
        self.reading = reading
        self.unit = unit
        self.timestamp = timestamp

schema_str = """{
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "title": "Temperature",
    "description": "Temperature sensor reading",
    "type": "object",
    "properties": {
      "city": {
        "description": "City name",
        "type": "string"
      },
      "reading": {
        "description": "Current temperature reading",
        "type": "number"
      },
      "unit": {
        "description": "Temperature unit (C/F)",
        "type": "string"
      },
      "timestamp": {
        "description": "Time of reading in ms since epoch",
        "type": "number"
      }
    }
  }"""

def dict_to_temp(dict, ctx):
    return Temperature(dict['city'], dict['reading'], dict['unit'], dict['timestamp'])
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/34d90b8f-01cb-4f4d-ab02-b8bef22e3b64.png)

#### Add Main Block

Databricks のノートブック上で下記のコードを記述します。

```python
if __name__ == '__main__':
    topic = 'temp_readings'

    json_deserializer = JSONDeserializer(schema_str, from_dict=dict_to_temp)
    set_consumer_configs()
    consumer = Consumer(config)
    consumer.subscribe([topic])
    while True:
        try:
            event = consumer.poll(1.0)
            if event is None:
                continue
            temp = json_deserializer(event.value(), 
                SerializationContext(topic, MessageField.VALUE))
            if temp is not None:
                print(f'Latest temp in {temp.city} is {temp.reading} {temp.unit}.')

        except KeyboardInterrupt:
            break

    consumer.close()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c8e1ce13-fb78-12b4-0eaf-ae7c79184ec0.png)

#### Execute the program

前の手順で記述したコードを実行します。処理が継続実行されるため、セルの実行をとめることを忘れないようでしてください。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/63f77b75-de9b-0f18-df30-9ba79e984874.png)

### Hands On: Use the Python AdminClient Class

#### Use AdminClient to Create a Topic and Alter its Configuration

Databricks のノートブック上で下記のコードを実行します。

```python
from confluent_kafka.admin import (AdminClient, NewTopic, 
                                   ConfigResource)
from config import config
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f9bf4483-70d0-00ab-d696-9365f16573cb.png)

#### Project Setup

Databricks 利用時には不要な手順であるため省略。

#### Add Required Imports

Databricks のノートブック上で下記のコードを実行します。

```python
# return True if topic exists and False if not
def topic_exists(admin, topic):
    metadata = admin.list_topics()
    for t in iter(metadata.topics.values()):
        if t.topic == topic:
            return True
    return False
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a198eb99-9971-9d29-cf51-aa90e867595e.png)


#### Check if Kafka Topic Exists

Databricks のノートブック上で下記のコードを実行します。

```python
# return True if topic exists and False if not
def topic_exists(admin, topic):
    metadata = admin.list_topics()
    for t in iter(metadata.topics.values()):
        if t.topic == topic:
            return True
    return False
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0e41ee49-d321-a9cf-fdda-55b0da21cfc5.png)

#### Create a New Kafka Topic

Databricks のノートブック上で下記のコードを実行します。

```python
# create new topic and return results dictionary
def create_topic(admin, topic):
    new_topic = NewTopic(topic, num_partitions=6, replication_factor=3) 
    result_dict = admin.create_topics([new_topic])
    for topic, future in result_dict.items():
        try:
            future.result()  # The result itself is None
            print("Topic {} created".format(topic))
        except Exception as e:
            print("Failed to create topic {}: {}".format(topic, e))
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4010bdf6-981c-11ac-7ba3-a27b0ae1ae44.png)

#### Describe the New Kafka Topic

Databricks のノートブック上で下記のコードを実行します。

```python
# get max.message.bytes property
def get_max_size(admin, topic):
    resource = ConfigResource('topic', topic)
    result_dict = admin.describe_configs([resource])
    config_entries = result_dict[resource].result()
    max_size = config_entries['max.message.bytes']
    return max_size.value
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/da5f8a6d-abdd-cfe0-c1f3-cd8e2898263b.png)

#### Set a Kafka Topic Configuration Property Value

Databricks のノートブック上で下記のコードを実行します。

```python
# set max.message.bytes for topic
def set_max_size(admin, topic, max_k):
    config_dict = {'max.message.bytes': str(max_k*1024)}
    resource = ConfigResource('topic', topic, config_dict)
    result_dict = admin.alter_configs([resource])
    result_dict[resource].result()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7a4a34ce-3c83-b989-8cdd-6272b05d4c5b.png)

#### Add Main Block

Databricks のノートブック上で下記のコードを記述します。

```python
if __name__ == '__main__':

    # Create Admin client
    admin = AdminClient(config)
    topic_name = 'my_topic'
    max_msg_k = 50

    # Create topic if it doesn't exist
    if not topic_exists(admin, topic_name):
        create_topic(admin, topic_name)

    # Check max.message.bytes config and set if needed
    current_max = get_max_size(admin, topic_name)
    if current_max != str(max_msg_k * 1024):
        print(f'Topic, {topic_name} max.message.bytes is {current_max}.')
        set_max_size(admin, topic_name, max_msg_k)

    # Verify config was set 
    new_max = get_max_size(admin, topic_name)
    print(f'Now max.message.bytes for topic {topic_name} is {new_max}')
```

#### Test admin.py

前の手順で記述したコードを実行します。私の環境ではエラーとなってしまいました。

```text
KafkaException: KafkaError{code=UNKNOWN_TOPIC_OR_PART,val=3,str="Broker: Unknown topic or partition"}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/600809ab-66bd-4746-4f9e-319b1550dd0d.png)

topic 作成後に1秒待機することで正常終了しました。

```python
if __name__ == '__main__':

    # Create Admin client
    admin = AdminClient(config)
    topic_name = 'my_topic'
    max_msg_k = 50

    # Create topic if it doesn't exist
    if not topic_exists(admin, topic_name):
        create_topic(admin, topic_name)

    # TOPIC が作成されるまで1秒待機
    time.sleep(1)

    # Check max.message.bytes config and set if needed
    current_max = get_max_size(admin, topic_name)
    if current_max != str(max_msg_k * 1024):
        print(f'Topic, {topic_name} max.message.bytes is {current_max}.')
        set_max_size(admin, topic_name, max_msg_k)

    # Verify config was set 
    new_max = get_max_size(admin, topic_name)
    print(f'Now max.message.bytes for topic {topic_name} is {new_max}')
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/22862139-d028-abef-59d6-f1b3c9273a2c.png)

#### Exercise Environment Teardown

本記事で作成した`learn-kafka-python` environment を削除します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b62e2a09-f76a-68fb-1dd3-18fc3452da41.png)