重要

このページの日本語コンテンツは古くなっている可能性があります。最新の英語版コンテンツをご覧になるには、こちらをクリックしてください。

Confluent Cloud の Schema Linking

Looking for Confluent Platform Schema Management docs? You are currently viewing Confluent Cloud documentation. If you are looking for Confluent Platform docs, check out Schema Management on Confluent Platform.

スキーマレジストリ で Schema Linking をサポートするようになりました。以下のビデオとドキュメントでは、お使いの Confluent デプロイで Schema Linking の使用を開始するために必要な概念、構成、ツールについて説明します。クイックスタートでは、エクスポーターを作成して、クラスターへの Schema Linking の実装に使用する方法を実践的な例で順を追って説明します。

ちなみに

The video below focuses on Schema Linking on Confluent Cloud, but the feature is also available on Confluent Enterprise, with Confluent Platform specific setup, commands, and workflow, as described in the Confluent Platform Schema Linking Quick Start. Schema Linking on Confluent Cloud and on Confluent Platform share the core concepts and mechanisms of schema contexts, exporters, configuration options, exporter lifecycle, and REST APIs.

Schema Linking とは

Schema Linking は、2 つの スキーマレジストリ クラスター間でスキーマを同期させ続けるための機能です。Schema Linking を Cluster Linking と組み合わせると、2 つの スキーマレジストリ および Kafka クラスターの間でスキーマとトピックデータの両方を同期し続けることができます。

../_images/schema-linking.ja.png

スキーマレジストリ に、Schema Linking をサポートするための 2 つの新しい概念が導入されました。

  • スキーマコンテキスト - スキーマコンテキスト は スキーマレジストリ の独立したスコープを表し、1 つの スキーマレジストリ クラスター内に任意の数の個別の "サブレジストリ" を作成するために使用できます。各スキーマコンテキストはスキーマ ID とサブジェクト名を個別にグループ化したものであり、同じスキーマ ID を異なるコンテキストで使用できるようにすることで、完全に異なるスキーマを表します。明示的なコンテキストがないスキーマ ID またはサブジェクト名はすべてデフォルトのコンテキストに存在し、1 つのドット(.)で示されます。明示的なコンテキストはドットで始まり、さらに別のドットで区切られた部分から構成されます(例: .mycontext.subcontext)。コンテキスト名の構成は Unix の絶対パスと似ていますが、フォワードスラッシュの代わりにドットが使用されます(デフォルトのスキーマは Unix のルートパスと似ています)。ただし、プレフィックスを共有する 2 つのコンテキストの間に関係はありません。
  • スキーマエクスポーター - スキーマエクスポーター は スキーマレジストリ に存在するコンポーネントであり、スキーマを 1 つの スキーマレジストリ クラスターから別のクラスターにエクスポートします。スキーマエクスポーターのライフサイクルは API を介して管理されます。これらの API で、スキーマエクスポーターの作成、一時停止、再開、破棄が可能です。スキーマエクスポーターは、スキーマの変更データキャプチャーを実行できる "小型コネクター" のようなものです。

以下のクイックスタートでは、Schema Linking を実現するためにスキーマエクスポーターとスキーマコンテキストの使用を開始する方法を説明します。

これらの概念の詳細については、「スキーマコンテキスト」および「スキーマエクスポーター」を参照してください。

クイックスタート

今すぐに Schema Linking を試してみたい場合は、以下の手順に従ってください。クイックスタートの後で、コンテキスト、エクスポーター、コマンドオプション、API に関する詳細情報を紹介しています。これらの情報は、いくつかの実践的な例を試した後の方が役立つと考えられます。

Confluent CLI の最新バージョンの取得

  • 既に Confluent CLI がある場合は、最新の状態であることを確認します。

    既に Confluent Cloud CLI がある場合は、ccloud update --major を実行すると、利用可能なアップデートと confluent v2.0 への直接アップグレードのプロンプトが表示されます。

    既に新しい統合 CLI がある場合は confluent update を実行します。統合 CLI を使用すると、コマンドで ccloud ではなく confluent を使用できます。アップグレードしたら、参考として 便利な統合 CLI の Confluent コマンドリファレンス をご覧ください。

  • 新しい統合 CLI と移行パスの詳細については、「Confluent CLI のインストール」、「Confluent CLI v2 への移行」、「複数の CLI の並列実行」を参照してください。

  • 最新バージョンの CLI では次の Schema Linking 機能を利用できます。

    New Features
    -------------
      - Support for passing Schema Registry API key and secret in produce/consume commands
      - New Schema linking commands under ``confluent schema-registry exporter``
    
  • ターミナルで直接エクスポーターの CLI に関するヘルプを表示するには、confluent schema-registry exporter と入力します。

送信元環境と送信先環境に 2 つの スキーマレジストリ クラスターをセットアップする

  1. Confluent Cloud にログオンし、それぞれ "SOURCE" と "DESTINATION" という名前の 2 つの新しい環境を作成します。

    Environments (右上のメニュー)に移動し、Add cloud environment をクリックして表示される指示に従います。

  2. 各環境で スキーマレジストリ を有効にできるように、Kafka クラスターを各環境に追加します(例: SOURCE 環境には "my-cluster-on-source"、DESTINATION 環境には "my-cluster-on-destination")。クラスタータイプ は問いません。

    この処理は、Confluent CLI または Confluent Cloud Console で行います。

    • Cloud Console の場合、1 つの環境で Add cluster をクリックし、指示に従ってクラスター名を指定してクラスターを作成します。

    • CLI の場合、confluent environment listconfluent environment use <environment-id> の順に入力して各環境に移動してから、該当する環境でクラスターを作成します。

      たとえば、SOURCE 環境では以下のようになります。

      confluent kafka cluster create my-cluster-on-source --type basic --cloud gcp --region us-east1
      

      DESTINATION 環境では以下のようになります。

      confluent kafka cluster create my-cluster-on-destination --type basic --cloud gcp --region us-west1
      
  3. 各環境で スキーマレジストリ をセットアップし、その API エンドポイントを確認します。

    • 各環境で Cloud Console の Schema Registry タブに移動し、Set up my own をクリックして表示される指示に従います。レジストリが作成されると Schema Registry タブが表示されるはずです。
    • 送信元と送信先両方の Schema Registry タブで、それぞれの API エンドポイントをコピーして保存します。
  4. 送信元と送信先の両方のクラスターで スキーマレジストリ クラスターを操作するための API キーとシークレットを作成します。

    Schema Registry タブに移動し、API credentials の下にある編集アイコンをクリックして表示される指示に従います。送信元と送信先両方のキーとシークレットのペアを必ず保存してください。後からシークレットを取得することはできません。

  5. 送信元と送信先のアクセス情報を確認します。

    この段階で、送信元と送信先のクラスター両方の スキーマレジストリ へのアクセスに関する詳細情報が 3 種類あるはずです。これらの情報は、この後の手順で使用します。

    • スキーマレジストリ URL(API エンドポイント)
    • API キー
    • API シークレット

    ちなみに

    送信元と送信先の URL(エンドポイント)は物理的な スキーマレジストリ クラスターを指定します。スキーマレジストリが同じ物理クラスター上にある場合、同じ URL を使用できます。これらは論理クラスターとしては別のものであり、その API キー/シークレットのペアは異なります。レジストリが異なる物理クラスター上にある場合、URL は異なるものになります。

送信元でスキーマを作成する

送信元環境では、少なくとも 2 つまたは 3 つのスキーマを作成します。そのうち 1 つ以上に 修飾サブジェクト名 を指定します。

Cloud Console から各スキーマを作成するには、以下の手順に従います。

  1. SOURCE 環境の Schema Registry タブで View and manage schemasAdd schema の順にクリックします。

  2. サブジェクト名を入力します。

    • 非修飾サブジェクト名でスキーマを作成するには、単純に coffeedonuts などの名前を指定します。
    • 特定のコンテキストにおいて修飾サブジェクト名でスキーマを作成するには、:.<context-name>:<subject-name> という構文を使用します。たとえば、:.snowcones:sales:.burgers:locations のように指定します。
    ../_images/schema-link-qualified-subject.png
  3. どちらの場合も、デフォルトのスキーマを使用します。

    このクイックスタートでは、スキーマそのものを操作するのではなく、サブジェクト名とコンテキストの使用方法を学びます。このため、時間の節約の目的でデフォルトのスキーマを使用しますが、必要に応じて特定のスキーマを使用しても構いません。

  4. Create をクリックします。

エクスポーターの認証情報を作成する

スキーマエクスポーターは送信元環境のスキーマを読み取り、リンクされたコピーを送信先にエクスポートします。このため、送信先にアクセスするための認証情報が必要です。

エクスポーターを作成するために使用する ~/config.txt を作成し、エクスポーターが DESTINATION クラスターにアクセスするために必要な URL と認証情報を入力します。

schema.registry.url=<destination sr url>
basic.auth.credentials.source=USER_INFO
basic.auth.user.info=<destination api key>:<destination api secret>

ちなみに

この認証情報は、エクスポーターが送信先にアクセスするために使用するものです。これらの環境でコマンドを実行するには、さらに送信元と送信先の認証情報が必要になります。詳細については、以降の該当箇所で説明します。

送信元でエクスポーターを作成する

  1. このクイックスタートの例では、送信元でエクスポーターを作成します。このため、現在の環境が SOURCE であることを確認してください。

    移動するには、confluent environment listconfluent environment use <environment-id> を使用します。

  2. confluent schema-registry exporter create コマンドを使用して新しいエクスポーターを作成します。

    このデモでは、特定のコンテキスト(デフォルト以外)のものを含め、すべてのスキーマをエクスポーターでコピーします。このため、--subjects フラグを「コンテキストワイルドカード」に含め、すべてのコンテキストのサブジェクトを指定します(--subjects ":*:")。

    confluent schema-registry exporter create <exporter-name> --subjects ":*:" --config-file ~/config.txt
    

    たとえば、以下のコマンドでは、特定のコンテキストとデフォルトのコンテキストのものを含む、すべてのスキーマをエクスポートする "my-first-exporter" というエクスポーターが作成されます。

    confluent schema-registry exporter create my-first-exporter --subjects ":*:" --config-file ~/config.txt
    
  3. 入力を求められたら、送信元の API キーとシークレットを入力します。

    Created schema exporter "my-first-exporter" という、エクスポーターが作成されたことを確認する出力が表示されます。

エクスポーターのその他のオプション

ここで作成したエクスポーターは、単にすべてをエクスポートするという意味で、比較的ベーシックなものです。次のセクションでわかるように、修飾サブジェクト名と非修飾サブジェクト名でスキーマを整理、エクスポート、操作する方法を理解するには、このようなエクスポーターが効率的です。

ただし、特定のサブジェクトとコンテキストのみをエクスポートするよう指定するエクスポーターも作成できることを覚えておいてください。その場合、以下の構文を使用します。

confluent schema-registry exporter create <exporterName> --subjects <subjectName1>,<subjectName2> \
--context-type CUSTOM --context-name <contextName> \
--config-file ~/config.txt
  • <> で囲まれた文字列は任意の名前に置き換えます。
  • subjects はコンマ区切りの文字列としてリストされます("pizzas,sales,customers" など)。
  • subjectscontext-typecontext-name はすべて省略可能です。context-type が CUSTOM の場合のみ context-name を指定します。
  • subjects のデフォルトは * で、context-type のデフォルトは AUTO です。

また、エクスポーターを作成するときにすべてをデフォルトとし、--subjects を指定しない場合、デフォルトのコンテキストのスキーマのみをエクスポートするエクスポーターが作成されます。

confluent schema-registry exporter create my-first-exporter --config-file ~/config.txt

このタイプのエクスポーターでは、修飾サブジェクト名を持つ送信元のスキーマが送信先にエクスポートされません。

confluent schema-registry exporter createconfluent schema-registry exporter update で使用できるもう 1 つの省略可能なパラメーターは --subject-format です。これは、送信先クラスターのサブジェクト名のフォーマットを指定します。デフォルトサブジェクト名と置き換えるプレースホルダーとして、${subject} を含めることができます。たとえば、サブジェクト orders の場合、dc_${subject} は、送信先サブジェクト名 dc_orders にマッピングされます。

複数のエクスポーターを同時に作成して実行できるため、このクイックスタートを最後まで終えたら、次は異なるパラメーターを指定したその他のエクスポーターを作成し、テストしてみてください。

スキーマエクスポーターのパラメーターの詳細については、「構成オプション」を参照してください。

エクスポーターが実行されていることを確認しエクスポーターの情報を表示する

引き続き SOURCE 環境で、以下のコマンドを実行します。

  1. 利用可能なエクスポーターのリストを表示します。

    confluent schema-registry exporter list
    

    作成したエクスポーターがリストに表示されます。

  2. エクスポーターに関する情報を表示します。

    confluent schema-registry exporter describe <exporterName>
    

    出力は以下のようになります。

    confluent schema-registry exporter describe my-first-exporter
    +--------------------------------+--------------------------------------------------------------------------+
    | Name                           | my-first-exporter                                                        |
    | Subjects                       | *                                                                        |
    | Context Type                   | AUTO                                                                     |
    | Context                        |                                                                        . |
    | Remote Schema Registry Configs | basic.auth.credentials.source="USER_INFO"                                |
    |                                | basic.auth.user.info="[hidden]"                                          |
    |                                | schema.registry.url="https://psrc-0xx5p.us-central1.gcp.confluent.cloud" |
    +--------------------------------+--------------------------------------------------------------------------+
    
  3. エクスポーターの構成を取得します。

    confluent schema-registry exporter get-config <exporterName>
    
  4. エクスポーターのステータスを取得します。

    confluent schema-registry exporter get-status <exporterName>
    
  5. 最後に確認として、送信元のスキーマのリストを取得します。

    すべてのスキーマのリストを表示するには、プレフィックスワイルドカードを使用します。

    confluent schema-registry subject list --prefix ":*:"
    

    このワイルドカードを使用する場合、実質的にコマンド confluent schema-registry subject list と同じ動作になります。このコマンドでは、送信元で作成したサブジェクトのリストが以下のように返されます。

            Subject
    +---------------------+
      :.burgers:locations
      :.snowcones:sales
      coffee
      customers
      donuts
    

スキーマがエクスポートされたことを確認する

ここまでで、エクスポーターが実行中であることを確認しました。自分が送信元で作成したスキーマについては把握できているため、自分のスキーマが送信先にエクスポートされていることを確認します。

  1. DESTINATION に切り替えます。

    移動するには、confluent environment listconfluent environment use <environment-id> を使用します。

  2. 以下のコマンドを実行してすべてのスキーマを表示します。入力を求められたら、DESTINATION の API キーとシークレットを入力します。

    confluent schema-registry subject list --prefix ":*:"
    

    クエリの実行時に、コンテキストがどのようにサブジェクトのプレフィックスとして渡されるかに注目してください。前述の例では、コンテキストワイルドカード :*: を使用してすべてのコンテキストに一致させています。これは実質的に汎用コマンド confluent schema-registry subject list と同じですが、特定のコンテキストを渡す場合は、より明確なサブジェクトのプレフィックスを使用する必要があります。

    出力される DESTINATION のスキーマのリストが SOURCE のリストと一致していなければなりません。

              Subject
    +--------------------------------+
      :.lsrc-3qy52.burgers:locations
      :.lsrc-3qy52.snowcones:sales
      :.lsrc-3qy52:coffee
      :.lsrc-3qy52:customers
      :.lsrc-3qy52:donuts
    
  3. 特定のコンテキストのスキーマのみのリストを表示します。

    プレフィックスを使用して送信先の全サブジェクトのリストを取得したら(前述の例を参照)、コンテキスト名のみを渡し、特定のコンテキストで絞り込んだサブジェクトのリストを表示できます。

    • たとえば、burgers コンテキストのスキーマのリストを表示するには、以下のように指定します。

      confluent schema-registry subject list --prefix ":.lsrc-3qy52.burgers:"
      

      出力は以下のようになります。

                   Subject
      +--------------------------------+
        :.lsrc-3qy52.burgers:locations
      
    • snowcones コンテキストのスキーマのリストを表示するには、以下のように指定します。

      confluent schema-registry subject list --prefix ":.lsrc-3qy52.snowcones:"
      

      出力は以下のようになります。

                   Subject
      +------------------------------+
        :.lsrc-3qy52.snowcones:sales
      

ちなみに

  • 送信元でエクスポーターを作成したときに省略可能なパラメーター --subject-format を使用した場合は、送信先にエクスポートされたサブジェクトが、指定したサブジェクト名変更フォーマットにマッピングされていることを確認します。

  • 同じ curl コマンドを使用して API を呼び出します。以下に例を示します。

    curl -u <destination api key>:<destination api secret> '<destination sr url>/subjects?subjectPrefix=:.context1:foo'
    

エクスポーターを一時停止して変更を加える

  1. エクスポーターを一時停止します。

    SOURCE に戻り、以下のコマンドを実行してエクスポーターを一時停止します。

    confluent schema-registry exporter pause <exporterName>
    

    コマンドが正常に実行されたことを確認するメッセージが出力されます。たとえば、Paused schema exporter "my-first-exporter" のようなメッセージです。

    念のためステータスを確認します。

    confluent schema-registry exporter get-status <exporterName>
    

    出力は以下のようになります。

    confluent schema-registry exporter get-status my-first-exporter
    +--------------------+-------------------+
    | Name               | my-first-exporter |
    | Exporter State     | PAUSED            |
    | Exporter Offset    |          10011386 |
    | Exporter Timestamp |     1631107710822 |
    | Error Trace        |                   |
    +--------------------+-------------------+
    
  2. スキーマエクスポーターのオフセットをリセットしてからステータスを取得します。

    confluent schema-registry exporter reset <exporterName>
    
    confluent schema-registry exporter get-status <exporterName>
    

    ステータスに、オフセットがリセットされたことが示されます。その例を次に示します。

    confluent schema-registry exporter get-status my-first-exporter
    +--------------------+-------------------+
    | Name               | my-first-exporter |
    | Exporter State     | PAUSED            |
    | Exporter Offset    |                -1 |
    | Exporter Timestamp |                 0 |
    | Error Trace        |                   |
    +--------------------+-------------------+
    
  3. エクスポーターの構成または情報をアップデートします。

    アップデートの対象を subjectscontext-typecontext-nameconfig-file から選択します。その例を次に示します。

    confluent schema-registry exporter update <exporterName> --context-name <newContextName>
    
  4. スキーマエクスポーターを再開します。

    confluent schema-registry exporter resume <exporterName>
    

エクスポーターを削除する

テストを終了できる状態になったら、以下のコマンドを使用してエクスポーターを一時停止し、削除します。

  1. エクスポーターを一時停止します。

    confluent schema-registry exporter pause <exporterName>
    
  2. エクスポーターを削除します。

    confluent schema-registry exporter delete <exporterName>
    

クイックスタートはこれで終了です。次のセクションでは、Schema Linking の概念と、ここで試したツールの詳細について説明します。

スキーマコンテキスト

スキーマコンテキストとは

スキーマコンテキストは単に「コンテキスト」とも呼ばれ、基本的には、サブジェクト名とスキーマ ID をグループ化したものです。1 つの スキーマレジストリ クラスターで任意の数のコンテキストをホストできます。各コンテキストは個別の "サブレジストリ" とみなすことができます。コンテキストは、スキーマエクスポーター を使用して別の スキーマレジストリ にコピーすることもできます。

コンテキストの仕組み

以下で、コンテキストの主要な側面をいくつか紹介し、それらがスキーマの整理にどのように役立つかについて説明します。

スキーマのスコープはコンテキストごとに指定される

サブジェクト名とスキーマ ID のスコープはコンテキストごとに指定されるため、同じ スキーマレジストリ クラスター内の 2 つのコンテキストにそれぞれ同じ ID(123 など)のスキーマ、または同じ名前(mytopic-value)のサブジェクトがあっても問題ありません。

デフォルトのコンテキスト

明示的なコンテキストがないスキーマ ID またはサブジェクト名はすべてデフォルトのコンテキストに存在し、1 つのドット(.)で表されます。明示的なコンテキストはドットで始まり、さらに別のドットで区切られた部分から構成されます(例: .mycontext.subcontext)。コンテキスト名は Unix の絶対パスと似ていますが、フォワードスラッシュの代わりにドットが使用されます(同様に、デフォルトのスキーマコンテキストは Unix のルートパスと似ています)。ただし、プレフィックスを共有する 2 つのコンテキストの間に関係はありません。

修飾サブジェクト

サブジェクト名がコンテキストで修飾されている場合、そのサブジェクトを "修飾サブジェクト" と呼びます。コンテキストでサブジェクトを修飾する場合、そのコンテキストをコロンで囲む必要があります。たとえば、:.mycontext:mysubject のようになります。修飾されていないサブジェクト名は、デフォルトのコンテキストにあるとみなされます。つまり、mysubject:.:mysubject と同じです(ドットはデフォルトのコンテキストを表します)。

コンテキストを REST API に渡す方法は 2 つあります。

要件を満たしたサブジェクトの使用

修飾サブジェクトは、サブジェクト名が想定される場所であればどこにでも渡すことができます。ほとんどの REST API は、POST /subjects/{subject}/versions のように、サブジェクト名を使用します。

URL パスの一部としてサブジェクト名を使用しない REST API はごくわずかです。その例を以下に示します。

  • /schemas/ids/{id}
  • /schemas/ids/{id}/subjects
  • /schemas/ids/{id}/versions

前述の 3 つの API で "subject"(?subject と記述します)というクエリパラメーターを使用できるようになったため、/schemas/ids/{id}?subject=:.mycontext:mysubject のように修飾サブジェクト名を渡すと、指定したコンテキストを使用したスキーマ ID のルックアップが実行されます。

ベースコンテキストパスの使用

前述したように、非修飾サブジェクトを指定するすべての API はデフォルトのコンテキストで機能します。サブジェクト名が想定される場所に修飾サブジェクトを渡す方法に加えて、ベースコンテキストパスを使用してコンテキストを渡す方法もあります。ベースコンテキストパスは /contexts/{context} という形式で、既存の スキーマレジストリ パスすべての先頭に追加できます。このため、特定のコンテキストでスキーマ ID をルックアップするために、URL /contexts/.mycontext/schemas/ids/{id} を使用することもできます。

ベースコンテキストパスを使用してデフォルトのコンテキストを操作することもできます。この場合、ベースコンテキストパスは "/contexts/:.:/" という形式になります(例: /contexts/:.:/schemas/ids/{id})。一部の URL パーサーで削除されてしまうため、1 つのドットは使用できません。

マルチコンテキスト API

これまで挙げたすべての例は 1 つのコンテキストに該当するものです。複数のコンテキスト向けに結果を返す API は 3 つあります。

  • /contexts
  • /schemas?subjectPrefix=:*:
  • /subjects?subjectPrefix=:*:

最初の API、/contexts は、すべてのコンテキストのリストを返します。他の 2 つの API、/schemas および /subjects は通常、デフォルトのコンテキストのみで動作します。これらを使用して、subjectPrefix を値 :*: (コンテキストワイルドカードと呼ばれます)とともに渡すことで、すべてのコンテキストに対してクエリを実行できます。コンテキストワイルドカードはすべてのコンテキストに一致します。

クライアント向けのコンテキスト名の指定

クライアントを使用して スキーマレジストリ と通信するときに、クライアントで特定のコンテキストを使用することもできます。たとえば、ある スキーマレジストリ との通信から別の Schema Registry との通信にクライアントを移行する場合などです。このために、上で定義したベースコンテキストパスを使用します。この場合、クライアントが使用する スキーマレジストリ の URL を https://<host1> から https://<host2>/contexts/.mycontext に変更するだけです。

スキーマレジストリ の URL でベースコンテキストパスを使用することで、クライアントがすべての スキーマレジストリ リクエストに同じスキーマコンテキストを使用するようになります。しかし、さらに高度な用法では、クライアントがトピックごとに異なるコンテキストを使用するようにできます。このためには、コンテキスト名戦略をシリアライザーまたはデシリアライザーに指定します。

  • context.name.strategy=com.acme.MyContextNameStrategy

コンテキスト名戦略は、以下のインターフェイスを実装する必要があるクラスです。

/**
 * A {@link ContextNameStrategy} is used by a serializer or deserializer to determine
 * the context name used with the schema registry.
 */
public interface ContextNameStrategy extends Configurable {

  /**
   * For a given topic, returns the context name to use.
   *
   * @param topic The Kafka topic name.
   * @return The context name to use
   */
  String contextName(String topic);
}

コンテキスト名戦略の使用は、一般的なものではありません。ベースコンテキストパスを スキーマレジストリ の URL に指定する方法でほとんどのニーズを満たすことができます。

スキーマエクスポーター

スキーマエクスポーターとは

以前は、Confluent Replicator が、ある スキーマレジストリ クラスターから別のクラスターに スキーマを移行 するための主要な手段でした。ただし、送信元の スキーマレジストリ クラスターがオンプレミスである必要がありました。この手法によるスキーマの移行をサポートするには、グローバルで、または特定のサブジェクトについて、送信先の スキーマレジストリ を IMPORT モードにします。

これに代わって新しいスキーマエクスポーター機能が導入され、Replicator のスキーマ移行機能が拡張されました。スキーマエクスポーターは スキーマレジストリ クラスター内に常駐し、Confluent Cloud の 2 つの スキーマレジストリ クラスター間でスキーマをレプリケートするために使用できます。

Schema Linking

スキーマエクスポーターにより、コンテキスト/修飾サブジェクト名を使用してレジストリ間のスキームを同期する Schema Linking に対応できます。スキーマコンテキストが概念的な基盤と名前空間フレームワークを提供する一方、エクスポーターは負担の大きなリンク処理を担います。

送信元のデフォルトのコンテキストから送信先の新しいコンテキストへのスキーマのエクスポート

デフォルトで、スキーマエクスポーターはスキーマを送信元の スキーマレジストリ のデフォルトのコンテキストから送信先の スキーマレジストリ の新しいコンテキストにエクスポートします。送信先のコンテキスト(つまり、送信先のコンテキスト内のサブジェクト)は IMPORT モードになります。これにより、送信先の スキーマレジストリ は、デフォルトのコンテキストのクライアントに一切影響を与えずに、デフォルトのコンテキストを通常どおりに使用できます。

送信先の スキーマレジストリ にデフォルトで作成される新しいコンテキストは .lsrc-xxxxxx という形式になり、送信元の論理名から取得されます。

スキーマレジストリ クラスターは互いにスキーマをエクスポートできる

2 つの スキーマレジストリ クラスターそれぞれでスキーマエクスポーターを使用して、デフォルトのコンテキストからもう一方の スキーマレジストリ にスキーマをエクスポートできます。この構成では、それぞれがデフォルトのコンテキストからの読み取りまたはデフォルトのコンテキストへの書き込みを実行でき、それぞれがエクスポートされたコンテキストからの読み取りを実行できます(ただし、書き込みは実行できません)。これにより、それぞれに送信元トピックと読み取り専用のミラートピックを持つ Cluster Linking の構成に一致させることができます。

エクスポーターは同じ スキーマレジストリ のコンテキスト全体にスキーマをコピーできる

さらに、スキーマエクスポーターは同じ スキーマレジストリ クラスター内のコンテキスト間でスキーマをコピーできます。たとえば、".staging" コンテキストを作成し、本稼働環境で使用する準備ができた段階で、そのスキーマを ".staging" コンテキストからデフォルトのコンテキストにコピーできます。同じ スキーマレジストリ クラスターとの間でスキーマのコピーを実行する場合は、特殊な URL local:/// を使用します。

スキーマのエクスポートのカスタマイズ

送信元の スキーマレジストリ からエクスポートされるコンテキストや送信先の スキーマレジストリ で使用されるコンテキストをカスタマイズするためのさまざまな方法が用意されています。全構成プロパティのリストについては、以下を参照してください。

Schema Registry ごとに使用できるエクスポーターの数

スキーマレジストリ ごとに同時に使用できるエクスポーターの数は 10 に制限されています。

構成オプション

スキーマエクスポーターの主要な構成プロパティは以下のとおりです。

name
エクスポーターの一意の名前。
subjects

複数の形式があります。

  • サブジェクト名/コンテキストのリスト(例: [ "subject1", "subject2", ".mycontext1", ".mycontext2" ]
  • サブジェクト名のプレフィックスを含み、ワイルドカードで終わるシングルトンのリスト(例: ["mytopic*"]
  • ワイルドカード 1 つのみを含むシングルトンのリスト["*"]。デフォルトのコンテキストのすべてのサブジェクトを示します。これがデフォルトです。
  • コンテキストワイルドカードを含むシングルトンのリスト、[":*:"]。すべてのコンテキストを示します。
subject-format
これはオプションのパラメーターで、送信先クラスターのサブジェクト名のフォーマットを指定するために使用できます。プレースホルダーとして ${subject} を指定できます。これは、デフォルトのサブジェクト名と置き換えられます。たとえば、サブジェクト orders の場合、dc_${subject} は、送信先サブジェクト名 dc_orders にマッピングされます。
context-type

以下のいずれかです。

  • AUTO - 送信元のコンテキストの先頭に自動生成されたコンテキスト(Confluent Cloud の場合 .lsrc-xxxxxx)を追加します。これがデフォルトです。
  • CUSTOM - 送信元のコンテキストの先頭に、以下の context で指定するカスタムのコンテキスト名を追加します。
  • NONE - 先頭に何も追加せずに、送信元のコンテキストをそのままコピーします。このオプションは、送信元 スキーマレジストリ そのものを送信先にコピーする場合に便利です。
context-name
前述の CUSTOM contextType で使用するコンテキスト名。
config

送信先 スキーマレジストリ と通信するクライアントを作成するための構成セット。構成ファイルで渡すことができます(例: --config-file ~/<my-config>.txt)。一般に、以下の要素が含まれます。

  • schema.registry.url - 送信先 スキーマレジストリ の URL。送信元と送信先が同じ場合、これを local:/// にすると、さらに効率的にコピーできます。
  • basic.auth.credentials.source - 通常は "USER_INFO"
  • basic.auth.user.info - 通常の形式は <api-key>:<api-secret>

System Topics and Security Configurations

The following configurations for system topics are available:

  • exporter.config.topic - Stores configurations for the exporters. The default name for this topic is _exporter_configs.
  • exporter.state.topic - Stores the status of the exporters. The default name for this topic is _exporter_states.

If you are using ロールベースアクセス制御(RBAC), exporter.config.topic and exporter.state.topic require ResourceOwner on these topics, as does the _schemas internal topic. See also, Use Role-Based Access Control (RBAC) in Confluent Cloud and Configuring Role-Based Access Control for Schema Registry on Confluent Platform.

If you are configuring スキーマレジストリ on Confluent Platform using the Schema Registry Security Plugin, you must activate both the exporter and the Schema Registry security plugin by specifying both extension classes in the $CONFLUENT_HOME/etc/schema-registry/schema-registry.properties files:

resource.extension.class=io.confluent.kafka.schemaregistry.security.SchemaRegistrySecurityResourceExtension,io.confluent.schema.exporter.SchemaExporterResourceExtension

The configuration for the exporter resource extension class in the schema-registry.properties is described in Set up source and destination environments in Schema Linking on Confluent Platform.

ライフサイクルとステート

スキーマレジストリ はスキーマを Kafka トピックに保存します。スキーマエクスポーターは、そのトピックのオフセットを使用して進捗状況を判断します。

スキーマエクスポーターが作成されると、最初は STARTING ステートになります。このステートでは、既にトピックに書き込まれた該当するスキーマすべてを検出し、エクスポートします。登録済みのスキーマをエクスポートした後、エクスポーターは RUNNING ステートに入ります。このステートでは、新しいスキーマがあると通知され、該当するスキーマがあればエクスポートできます。スキーマのエクスポート中、エクスポーターは最新のトピックのオフセットを記録してその進捗状況を保存します。

スキーマエクスポーターに変更を加える場合、まず "一時停止" する必要があります。これによって、スキーマエクスポーターが PAUSED ステートに入ります。適宜変更を加えた後で、エクスポーターを再開できます。再開後、エクスポーターは記録されている最新のオフセットの後に該当するスキーマがあればそれを検出し、エクスポートします。

一時停止中のエクスポーターを "リセット" することもできます。これにより保存されているオフセットがクリアされ、再開時に該当するすべてのスキーマが再エクスポートされます。この場合、リセット後にエクスポーターが再び STARTING ステートから開始され、同じライフサイクルを進みます。

ライフサイクルの各段階におけるスキーマエクスポーターのステートを以下にまとめます。

ステート 説明
STARTING エクスポーターはトピックの該当する登録済みスキーマすべてを検出し、エクスポートします。これは開始時のステートまたはリセット後のステートです。
RUNNING エクスポーターは新しいスキーマを通知され、該当する場合はエクスポートして、最新のトピックのオフセットを記録することで進捗状況を追跡します。
PAUSED 構成の変更などを目的に、エクスポーターを一時停止できます。再開時に、エクスポーターは最新の記録済みオフセット以降のスキーマを検出し、エクスポートします。

REST API

スキーマレジストリ は以下の REST API をサポートしています。詳細については、Schema Registry API のドキュメント の「エクスポーター」を参照してください。

タスク API
テナントのエクスポーターのリストを取得する GET /exporters
新しいエクスポーターを作成する POST /exporters
エクスポーターに関する情報を取得する GET /exporters/{name}
エクスポーターの構成を取得する GET /exporters/{name}/config
エクスポーターのステータスを取得する GET /exporters/{name}/status
エクスポーターの情報をアップデートする PUT /exporters/{name}/config
エクスポーターを一時停止する PUT /exporters/{name}/pause
エクスポーターを再開する PUT /exporters/{name}/resume
エクスポーターをリセットし、オフセットをクリアする PUT /exporters/{name}/reset
エクスポーターを削除する DELETE /exporters/{name}