Salesforce Docs Portal

Confluent Schema Registry 1.0 の例

この例では、Confluent Schema Registry を Kafka Connector と統合することで avro メッセージをパブリッシュおよびコンシュームし、メッセージのサイズを縮小する方法を示します。この例には次の 2 つのフローが含まれます。

  1. 最初のフローでは、スキーマが含まれる avro メッセージを生成して Kafka トピックに送信します。​[Transform Message]​ コンポーネントではスキーマが含まれる avro メッセージを作成し、​[Replace AVRO schema with id]​ 操作ではメッセージに埋め込まれたスキーマを Confluent Schema Registry からの ID で置き換えます。

  2. 2 番目のフローでは、Kafka トピックから avro メッセージを読み取り、非シリアル化された avro メッセージのコンテンツをコンソールに出力します。​[Replace id with AVRO schema]​ 操作では、受信したメッセージ内の ID を、Confluent Schema Registry から取得した実際のスキーマに置き換えます。

始める前に

以下が必要です。

  • Java 8

  • Anypoint Studio 7.7.0 以降

  • Mule runtime Engine (Mule) 4.2.x 以降

  • DataWeave

プロデューサーフローを作成する

プロデューサーフローでは、設定済みの Kafka トピックにメッセージをパブリッシュします。次のスクリーンショットは、Anypoint Studio のプロデューサーフローを示しています。

Studio での Confluent Schema Registry Connector のプロデューサーフロー

このフローでは、次を設定します。

  • HTTP の ​[Listener]​ コンポーネント

  • [Transform Message]​ コンポーネント

  • [Logger]​ コンポーネント

  • [Replace AVRO schema with id]​ 操作

  • [Publish]​ 操作

  • 2 つ目の ​[Transform Message]​ コンポーネント

HTTP リスナーを設定する

localhost​ のポート ​8082​ で ​/​ アカウントパスへのコールが行われたときに Mule フローを開始するように ​HTTP Listener​ を設定します。

この例では一部の項目値に変数を使用しています。次のいずれかを実行できます。

  • コード内で変数を値に置き換える。

  • プロパティファイルで各変数の値を指定しておき、コネクタ設定からそのファイルを参照する。

プロパティファイルの使用方法についての詳細は、​「プロパティプレースホルダーの設定」​を参照してください。

HTTP リスナー​を設定するには、次の手順に従います。

  1. Studio で新しい Mule プロジェクトを作成します。

  2. [Mule Palette (Mule パレット)]​ で「​http​」を検索し、​[Listener]​ 操作を選択します。

  3. [Listener]​ 操作をキャンバスにドラッグします。

  4. [Listener]​ 設定で、​[Connector configuration (コネクタ設定)]​ 項目の横にある ​[Add (追加)]​ アイコンをクリックしてグローバル要素を追加します。

  5. [OK]​ をクリックして、デフォルトを受け入れます。

  6. [Path (パス)]​ 項目を ​/​ に設定します。

最初の [Transform Message] コンポーネントを追加する

最初の ​[Transform Message]​ コンポーネントでは、スキーマが含まれる avro メッセージを生成します。

  1. [Mule Palette (Mule パレット)]​ で、「​transform message​」を検索します。

  2. [Transform Message]​ コンポーネントをキャンバスの ​[Listener]​ コンポーネントの横にドラッグします。

  3. Transform Message​ 設定で、​[Output (出力)]​ セクションの括弧を次の XML で上書きします。

    %dw 2.0
output application/avro schemaUrl='classpath://schema.json'
---
{
	"f1": "demo_message"
}

この例で使用するスキーマのコンテンツを次に示します。

{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string"}]}

このスキーマでは、​f1​ という名前の項目と型 ​string​ が含まれる JSON オブジェクトを記述しています。

Logger コンポーネントを追加する

  1. [Mule Palette (Mule パレット)]​ で、「​logger​」を検索します。

  2. [Logger]​ コンポーネントをキャンバスの ​[Transform Message]​ コンポーネントの横にドラッグします。

  3. 次の項目を設定します。

    項目 値

    Display Name (表示名)

    ロガーの名前 (​Log Response​ など)

    Message (メッセージ)

    「​#[payload]​」と入力します。

    Level (レベル)

    INFO (Default) (INFO (デフォルト))

[Replace AVRO Schema with ID] 操作を追加する

[Replace AVRO schema with id]​ 操作では、avro メッセージ内のスキーマを Confluent Schema Registry からの ID に置き換えて、メッセージのサイズを縮小します。

  1. [Replace AVRO schema with id]​ 操作をキャンバスの ​[Listener]​ コンポーネントの横にドラッグします。

  2. [Replace AVRO schema with id]​ の設定で、​[Connector configuration (コネクタ設定)]​ ドロップダウンをクリックし、​[Confluent_Schema_Registry_Connector_Config]​ を選択します。

  3. [Replace AVRO schema with id]​ プロパティウィンドウで、次の項目を設定します。

    項目 値

    Subject (件名)

    ${config.subject}

    Message (メッセージ)

    payload

[Publish] 操作を追加する

[Publish]​ 操作では、Kafka トピックにメッセージをパブリッシュします。

  1. [Publish]​ 操作をキャンバスの ​[Replace AVRO schema with id]​ 操作の横にドラッグします。

  2. [Publish]​ プロパティウィンドウで、次の項目を設定します。

    項目 値

    Display Name (表示名)

    操作の名前 (​Producer​ など)

    Topic (トピック)

    #[payload.topic]

    Key (キー)

    #[now()]

  3. [Publish]​ 操作の ​[General (一般)]​ 設定画面で、​[Add (追加)]​ アイコンをクリックして、グローバル要素設定項目にアクセスします。

    1. [Bootstrap server URLs (ブートストラップサーバー URL)]​ 項目で、​[Edit inline (インライン編集)]​ を選択し、​[Add (追加)]​ アイコンをクリックします。

    2. 値 ​${config.basic.bootstrapServers}​ を入力して ​[Finish (完了)]​ をクリックします。

2 つ目の [Transform Message] コンポーネントを追加する

2 つ目の ​[Transform Message]​ コンポーネントでは、​[Publish]​ 操作の応答を出力します。

  1. [Mule Palette (Mule パレット)]​ で、「​transform message​」を検索します。

  2. [Transform Message]​ コンポーネントをキャンバスの ​[Publish]​ 操作の横にドラッグします。

  3. Transform Message​ 設定で、​[Output (出力)]​ セクションの括弧を次の XML で上書きします。

    %dw 2.0
output application/json
---
payload

コンシューマーフローを作成する

コンシューマーフローでは、Kafka トピックからパブリッシュ済みのメッセージをコンシュームしてコンソールに出力します。次のスクリーンショットは、Anypoint Studio のコンシューマーフローを示しています。

Studio での Confluent Schema Registry Connector のコンシューマーフロー

このフローでは、次を設定します。

  • [Message listener]

  • [Logger]​ コンポーネント

  • [Replace id with AVRO schema]​ 操作

  • [Transform Message]​ コンポーネント

  • 2 つ目の ​[Logger]​ コンポーネント

[Message Listener] を追加する

[Message listener]​ では、Kafka トピックからパブリッシュ済みのメッセージをコンシュームします。

  1. [Message listener]​ をキャンバスにドラッグします。

  2. [Message listener]​ プロパティウィンドウで、次の項目を設定します。

    項目 値

    Display Name (表示名)

    操作の名前 (​Producer​ など)

    Topic (トピック)

    #[payload.topic]

    Key (キー)

    #[now()]

  3. [Connector configuration (コネクタ設定)]​ 項目の横にある ​[Add (追加)]​ アイコンをクリックし、グローバル要素設定項目にアクセスします。

  4. 次の項目に入力します。

    1. [Bootstrap server URLs (ブートストラップサーバー URL)]​ 項目で、​[Edit inline (インライン編集)]​ を選択し、​[Add (追加)]​ アイコンをクリックします。

    2. 値 ​${config.basic.bootstrapServers}​ を入力して ​[Finish (完了)]​ をクリックします。

    3. [Group ID (グループ ID)]​ 項目に ​${consumer.groupId}​ と入力します。

    4. [Topic Subscription Patterns (トピックサブスクリプションパターン)]​ 項目で、​[Edit inline (インライン編集)]​ を選択し、​[Add (追加)]​ アイコンをクリックします。

    5. 値 ​${config.topics}​ を入力して ​[Finish (完了)]​ をクリックします。

最初の [Logger] コンポーネントを追加する

  1. [Mule Palette (Mule パレット)]​ で、「​logger​」を検索します。

  2. [Logger]​ コンポーネントをキャンバスの ​[Message listener]​ の横にドラッグします。

  3. 次の項目を設定します。

    項目 値

    Display Name (表示名)

    ロガーの名前 (​Log Response​ など)

    Message (メッセージ)

    「​#[payload]​」と入力します。

    Level (レベル)

    INFO (Default) (INFO (デフォルト))

[Replace ID with AVRO Schema] 操作を追加する

[Replace id with AVRO schema]​ 操作では、埋め込まれた ID を avro スキーマに置き換えます。

  1. [Replace id with AVRO schema]​ 操作をキャンバスの ​[Logger]​ コンポーネントの横にドラッグします。

  2. [Replace id with AVRO schema]​ の設定で、​[Connector configuration (コネクタ設定)]​ ドロップダウンをクリックし、​[Confluent_Schema_Registry_Connector_Config]​ を選択します。

  3. [Replace id with AVRO schema]​ プロパティウィンドウで、次の項目を設定します。

    項目 値

    Message (メッセージ)

    payload

[Transform Message] コンポーネントを追加する

[Transform Message]​ コンポーネントでは、ペイロードを JSON に変換します。

  1. [Mule Palette (Mule パレット)]​ で、「​transform message​」を検索します。

  2. [Transform Message]​ コンポーネントをキャンバスの ​[Replace id with AVRO schema]​ 操作の横にドラッグします。

  3. Transform Message​ 設定で、​[Output (出力)]​ セクションの括弧を次の XML で上書きします。

    %dw 2.0
output application/json
---
payload

2 つ目の [Logger] コンポーネントを追加する

  1. [Mule Palette (Mule パレット)]​ で、「​logger​」を検索します。

  2. [Logger]​ コンポーネントをキャンバスの ​[Transform Message]​ の横にドラッグします。

  3. 次の項目を設定します。

    項目 値

    Display Name (表示名)

    ロガーの名前 (​Log Response​ など)

    Message (メッセージ)

    「​#[payload[0]]​」と入力します。

    Level (レベル)

    INFO (Default) (INFO (デフォルト))

Avro メッセージをパブリッシュおよびコンシュームするための XML

<?xml version="1.0" encoding="UTF-8"?>

<mule xmlns:ee="http://www.mulesoft.org/schema/mule/ee/core" xmlns:http="http://www.mulesoft.org/schema/mule/http"
	xmlns:kafka="http://www.mulesoft.org/schema/mule/kafka"
	xmlns:confluent-schema-registry="http://www.mulesoft.org/schema/mule/confluent-schema-registry" xmlns="http://www.mulesoft.org/schema/mule/core" xmlns:doc="http://www.mulesoft.org/schema/mule/documentation" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd
http://www.mulesoft.org/schema/mule/confluent-schema-registry http://www.mulesoft.org/schema/mule/confluent-schema-registry/current/mule-confluent-schema-registry.xsd
http://www.mulesoft.org/schema/mule/kafka http://www.mulesoft.org/schema/mule/kafka/current/mule-kafka.xsd
http://www.mulesoft.org/schema/mule/http http://www.mulesoft.org/schema/mule/http/current/mule-http.xsd
http://www.mulesoft.org/schema/mule/ee/core http://www.mulesoft.org/schema/mule/ee/core/current/mule-ee.xsd">
	<configuration-properties file="mule-app.properties"/>

	<http:listener-config name="HTTP_Listener_config" doc:name="HTTP Listener config" doc:id="ac9811ff-9234-4f44-9dc0-83100cb6c1bd" >
		<http:listener-connection host="0.0.0.0" port="8081" />
	</http:listener-config>
	<confluent-schema-registry:config name="Confluent_Schema_Registry_Connector_Config" doc:name="Confluent Schema Registry Connector Config" doc:id="faae820f-cd5a-46d0-9692-f28b53ea3bb6" >
		<confluent-schema-registry:basic-auth-connection username="${config.user}" password="${config.pass}" baseUri="${config.schemaUrl}" />
	</confluent-schema-registry:config>
	<kafka:consumer-config name="Apache_Kafka_Consumer_configuration" doc:name="Apache Kafka Consumer configuration" doc:id="d0443fc9-b2d3-4bbc-939f-126e98255cf3" >
		<kafka:consumer-plaintext-connection groupId="${config.consumerGroup}" >
			<kafka:bootstrap-servers >
				<kafka:bootstrap-server value="${config.bootstrapServer}" />
			</kafka:bootstrap-servers>
			<kafka:topic-patterns >
				<kafka:topic-pattern value="${config.topic}" />
			</kafka:topic-patterns>
		</kafka:consumer-plaintext-connection>
	</kafka:consumer-config>

	<kafka:producer-config name="Apache_Kafka_Producer_configuration" doc:name="Apache Kafka Producer configuration" doc:id="f378b3d4-4486-487d-84e2-8cc80aae7295" >
		<kafka:producer-plaintext-connection >
			<kafka:bootstrap-servers >
				<kafka:bootstrap-server value="${config.bootstrapServer}" />
			</kafka:bootstrap-servers>
		</kafka:producer-plaintext-connection>
	</kafka:producer-config>
	<flow name="demoFlow" doc:id="80c4cd43-2c65-4395-8e3e-52e6f7ed882b" >
		<http:listener doc:name="Listener" doc:id="a0c9691a-b202-4aaf-9dcf-6b1385709ee6" config-ref="HTTP_Listener_config" path="/test"/>
		<ee:transform doc:name="Transform Message" doc:id="4dc43b6d-f7c7-4cba-bbcd-db733cae1de4" >
			<ee:message >
				<ee:set-payload ><![CDATA[%dw 2.0
output application/avro schemaUrl='classpath://schema.json'
---
{
	"f1": "demo_message"
}]]></ee:set-payload>
			</ee:message>
		</ee:transform>
		<confluent-schema-registry:replace-avro-schema-with-id doc:name="Replace AVRO schema with id" doc:id="61b1d3b7-eabe-4d53-b5ca-357fda8bbd1e" config-ref="Confluent_Schema_Registry_Connector_Config" subject="${config.subject}"/>
		<kafka:publish doc:name="Publish" doc:id="aad92d55-a388-4680-86f1-e99adfcb14f5" config-ref="Apache_Kafka_Producer_configuration" topic="${config.topic}"/>
		<ee:transform doc:name="Transform Message" doc:id="f64a1dd7-d091-47bc-a0c7-b533c4c8d492" >
			<ee:message >
				<ee:set-payload ><![CDATA[%dw 2.0
output application/json
---
payload]]></ee:set-payload>
			</ee:message>
		</ee:transform>
	</flow>
	<flow name="demoFlow1" doc:id="a244ee60-143f-4320-a34e-6b572de50341" >
		<kafka:message-listener doc:name="Message listener" doc:id="056db197-1353-407a-ba58-8d90fc68e766" config-ref="Apache_Kafka_Consumer_configuration"/>
		<confluent-schema-registry:replace-id-with-avro-schema doc:name="Replace Id With Avro Schema" doc:id="26a15edf-94e2-4138-8acb-385821fc9710" config-ref="Confluent_Schema_Registry_Connector_Config"/>
		<ee:transform doc:name="Transform Message" doc:id="a17bd48c-37e7-449e-84a0-d8cd14e4925f" >
			<ee:message >
				<ee:set-payload ><![CDATA[%dw 2.0
output application/json
---
payload]]></ee:set-payload>
			</ee:message>
		</ee:transform>
	</flow>
</mule>

関連情報