How to Use Kafka Connect - Getting Started¶
This page includes the following topics and will help you in getting started with Kafka Connect. Confluent recommends you read and understand Kafka Connect Concepts before proceeding. Kafka Connect 101 is also a free course you can check out before moving ahead.
- デプロイに関する考慮事項
- Connect プラグインのインストール
- ワーカーの構成と実行
- キーコンバーターと値コンバーターの構成
- Connect プロデューサーおよびコンシューマー
- ソースコネクターのトピックの自動作成
- Connect Reporter
- ConfigProvider Interface
- Shutting Down Kafka Connect
- 次のステップ(追加のリファレンスとデモのリンク)
デプロイに関する考慮事項¶
To get started with Kafka Connect, you must have a set of Kafka brokers. The Kafka brokers can be earlier broker version, or the latest version. For more details, see Cross-Component Compatibility.
In addition to Kafka brokers, there are a few deployment options to consider as well. Understanding and acting on these deployment options ensures your Kafka Connect deployment will scale and support the long-term needs of your data pipeline.
Confluent Schema Registry¶
Although Schema Registry is not a required service for Kafka Connect, it enables you to easily use Avro, Protobuf, and JSON Schema as common data formats for the Kafka records that connectors read from and write to. This keeps the need to write custom code at a minimum and standardizes your data in a flexible format. You also get the added benefit of schema evolution and enforced compatibility rules. For more information, see Using Kafka Connect with Schema Registry and キーコンバーターと値コンバーターの構成.
スタンドアロンと分散モード¶
Connectors and tasks are logical units of work that run as a process. The process is called a worker in Kafka Connect. There are two modes for running workers:
Standalone mode: Useful for development and testing Kafka Connect on a local machine. It can also be used for environments that typically use single agents (for example, sending web server logs to Kafka).
Distributed mode: Runs Connect workers on multiple machines (nodes), which form a Connect cluster. Kafka Connect distributes running connectors across the cluster. You can add or remove nodes as your needs evolve.
This mode is also more fault tolerant. For example, if a node unexpectedly leaves the cluster, Kafka Connect distributes the work of that node to other nodes in the cluster. And, because Kafka Connect stores connector configurations, status, and offset information inside the Kafka cluster where it is safely replicated, losing the node where a Connect worker runs does not result in any loss of data.
重要
分散モードには拡張性、高可用性、および管理面での利点があるため、本稼働環境には分散モードが推奨されています。
Try to identify which mode works best for your environment before getting started. For more information about the different deployment modes for Kafka Connect workers check out this video.
Operating environment¶
Connect workers operate well in containers and managed environments, such as Kubernetes, Apache Mesos, Docker Swarm, or Yarn. The distributed worker stores all states in Kafka making it easier to manage a cluster. And, by design, Kafka Connect does not automatically handle restarting or scaling workers. This means your existing cluster management solution can continue to be used transparently. Note that the standalone worker state is stored on the local file system.
参考
- Docker の使用の詳細については、「Docker を使用したインストール」を参照してください。
- Kubernetes 環境での Confluent Platform のデプロイと管理の詳細については、「Kubernetes 用の Confluent」を参照してください。
Kafka Connect workers are JVM processes that can run on shared machines with sufficient resources. Hardware requirements for Connect workers are similar to that of standard Java producers and consumers. Resource requirements mainly depend on the types of connectors operated by the workers. More memory is required for environments where large messages are sent and where a large numbers of messages get buffered before being written in aggregate form to an external system. Using compression continuously requires a more powerful CPU.
If you have multiple workers running concurrently on a single machine, ensure you know the resource limits (CPU and memory). Start with the default heap size setting and monitor internal metrics and the system. Verify the CPU, memory, and network (10 GbE or greater) are sufficient for the load.
Connect プラグインのインストール¶
Kafka Connect is designed to be extensible so that developers can create custom connectors, transforms, and converters, and users can install and run them. This section will help you with installing Connect plugins.
Defining plugins¶
Kafka Connect プラグインは、1 つ以上のコネクター、変換、またはコンバーターの実装を含む JAR ファイルのセットです。Connect では各プラグインが互いに分離されるため、1 つのプラグインのライブラリが他のプラグインのライブラリから影響を受けることはありません。これは、複数のプロバイダーのコネクターを組み合わせて使用する場合に非常に重要なポイントです。
注意
It's common to have many plugins installed in a Connect deployment. Ensure you have only one version of each plugin installed.
A Kafka Connect plugin can be any one of the following:
- A directory on the file system that contains all required JAR files and third-party dependencies for the plugin. This is most common and is preferred.
- A single uber JAR containing all the class files for a plugin and its third-party dependencies.
A Kafka Connect plugin should never contain any libraries provided by the
Kafka Connect runtime. Kafka Connect finds the plugins using a plugin path
defined as a comma-separated list of directory paths in the plugin.path
worker configuration property. The
following shows an example plugin.path
worker configuration property:
plugin.path=/usr/local/share/kafka/plugins
Installing plugins¶
To install a plugin, you must place the plugin directory or uber JAR (or a
symbolic link that resolves to one of these) in a directory already listed in
the plugin path. Or, you can update the plugin path by adding the absolute path
of the directory containing the plugin. Using the plugin path example above, you
would create a /usr/local/share/kafka/plugins
directory on each machine
running Connect and then place the plugin directories (or uber JARs) there.
Connect ワーカーを起動すると、各ワーカーにより、プラグインパス上のディレクトリ内にあるすべてのコネクター、変換、およびコンバーターのプラグインが検出されます。コネクター、変換、またはコンバーターを使用する場合、Connect ワーカーにより、最初にそれぞれのプラグインからクラスが読み込まれ、その後 Kafka Connect ランタイムと Java ライブラリが読み込まれます。Connect では、他のプラグインに含まれているライブラリはすべて明示的に回避されます。これにより競合が防止され、さまざまなプロバイダーが開発したコネクターや変換の追加および使用が非常に簡単になります。
To find the components that fit your needs, check out the Confluent Hub page–it has an ecosystem of connectors, transforms, and converters. From Confluent Hub, you can easily install the components into your local Confluent Platform environment. For Confluent Hub Client installation instructions, see Confluent Hub Client. For a full list of supported connectors, see Supported Connectors.
Earlier versions of Kafka Connect required a different approach to installing
connectors, transforms, and converters. All the scripts for running Connect
recognized the CLASSPATH
environment variable. You would export this
variable to define the list of paths to the connector JAR files. The following
example shows an older CLASSPATH
export variable mechanism:
export CLASSPATH=/path/to/my/connectors/*
bin/connect-standalone standalone.properties new-custom-connector.properties
Confluent does not recommend exporting CLASSPATH
environment variable as
using this method to create a path to plugins can result in library conflicts
that can cause Kafka Connect and connectors to fail. Use the plugin.path
configuration property which properly isolates each plugin from other plugins
and libraries.
注釈
As described in Installing Connect Plugins, connector plugin JAR files
are placed in the plugin path (Connect worker
property: plugin.path
). However, a few connectors may require that you
additionally export the CLASSPATH
to the plugin JAR files when starting
the connector (export CLASSPATH=<path-to-jar-files>
). While not
recommended, CLASSPATH
is required for these connectors because
Kafka Connect uses classloading isolation to distinguish between system
classes and regular classes, and some plugins load system classes (for
example, javax.naming
and others in the package javax
). An example
error message showing this issue is provided below. If you see an error that
resembles the example below, in addition to adding the plugin path, you must also export CLASSPATH=<path-to-jar-files>
when starting the connector.
Caused by: javax.naming.NoInitialContextException:
Cannot instantiate class: com.tibco.tibjms.naming.TibjmsInitialContextFactory
[Root exception is java.lang.ClassNotFoundException: com.tibco.tibjms.naming.TibjmsInitialContextFactory]
ちなみに
By default, connectors inherit the partitioner used for the Kafka topic. You
can create a custom partitioner for a connector which you must place in the
connector’s /lib
folder.
You can also put partitioners in a common location of choice. If you choose
this option, you must add a symlink to the location from each connector’s
/lib
folder. For example, you would place a custom partitioner in the
path share/confluent-hub-components/partitioners
and then add the symlink
share/confluent-hub-components/kafka-connect-s3/lib/partitioners -> ../../partitioners
.
ワーカーの構成と実行¶
The following sections provide information about running workers in standalone mode and distributed mode. For a list of worker configuration properties, see Kafka Connect Worker Configuration Properties.
This video explains the different deployment modes for Kafka Connect workers.
スタンドアロンモード¶
Standalone mode is typically used for development and testing, or for lightweight, single-agent environments-for example, sending web server logs to Kafka. The following example shows a command that launches a worker in standalone mode:
bin/connect-standalone worker.properties connector1.properties [connector2.properties connector3.properties ...]
最初のパラメーター(worker.properties
)は、 ワーカーの構成プロパティファイル です。worker.properties
は、ファイル名の 1 つの例であることに注意してください。ワーカーの構成ファイルには、任意の有効なファイル名を使用できます。このファイルにより、使用する Kafka クラスターやシリアル化のフォーマットなどの設定を制御することができます。スタンドアロンモードで Avro と Schema Registry を使用するサンプル構成ファイルを参照するには、etc/schema-registry/connect-avro-standalone.properties
にあるファイルを開きます。このファイルをコピーして変更し、スタンドアロンワーカーのプロパティファイルとして使用することができます。
2 番目のパラメーター(connector1.properties
)は、コネクター構成プロパティファイルです。すべてのコネクターには、ワーカーで読み込まれる構成プロパティがあります。例に示されているように、このコマンドを使用して複数のコネクターを起動することができます。
同じホストマシン上で複数のスタンドアロンワーカーを実行する場合は、以下の 2 つの構成プロパティを各ワーカーで一意にする必要があります。
offset.storage.file.filename
: The storage file name for connector offsets. This file is stored on the local filesystem in standalone mode. Using the same file name for two workers will cause offset data to be deleted or overwritten with different values.listeners
: A list of URIs the REST API will listen on in the formatprotocol://host:port,protocol2://host2:port
–the protocol is either HTTP or HTTPS. You can specify hostname as0.0.0.0
to bind to all interfaces or leave hostname empty to bind to the default interface.
分散モード¶
分散モードには、ワーカー構成ファイルの読み込み以外の追加のコマンドラインパラメーターはありません。新しいワーカーは、新しいグループを開始するか、group.id
が一致する既存のグループに参加します。その後、ワーカーはコンシューマーグループと調整し、行うべき作業を分散させます。
分散モードでワーカーを起動するコマンドの例を以下に示します。
bin/connect-distributed worker.properties
Avro と Schema Registry を使用する分散モードの構成ファイルの例を参照するには、etc/schema-registry/connect-avro-distributed.properties
にあるファイルを開きます。このファイルのコピーを作成して変更を加え、新しい worker.properties
ファイルとして使用することができます。worker.properties
は、ファイル名の 1 つの例であることに注意してください。プロパティファイルには、任意の有効なファイル名を使用できます。
In standalone mode, connector configuration property files are added as commmand-line parameters. However, in distributed mode, connectors are deployed and managed using a REST API request. To create connectors, you start the worker and then make a REST request to create the connector. REST request examples are provided in many supported connector documents. For instance, see the Azure Blob Storage Source connector REST-based example for one example.
Note that if you run many distributed workers on one host machine for development
and testing, the listeners
configuration property must be unique for each
worker. This is the port the REST interface listens on for HTTP requests.
Connect の内部トピック¶
Connect では、コネクターとタスクの構成、オフセット、およびステータスがいくつかの Kafka トピックに保管されます。それらは、Kafka Connect の内部トピックと呼ばれます。この内部トピックには、高いレプリケーション係数、圧縮クリーンアップポリシー、および適切な数のパーティションを持たせることが重要です。
Kafka Connect では、起動時に内部トピックが自動的に作成されます。Connect の ワーカーの構成プロパティ を使用して、それらのトピックの名前、レプリケーション係数、およびパーティションの数を指定できます。Connect では、プロパティが要件を満たしていることが検証され、すべてのトピックが圧縮クリーンアップポリシーが有効な状態で作成されます。
Distributed workers that are configured with matching group.id
values
automatically discover each other and form a Kafka Connect cluster. All
workers in the cluster use the same three internal topics to share connector
configurations, offset data, and status updates. For this reason, all
distributed worker configurations in the same Connect cluster must have
matching config.storage.topic
, offset.storage.topic
, and
status.storage.topic
properties.
In addition to the three required internal topic names, the distributed worker configuration should have identical values for the following listed properties. This ensures that any worker in the cluster will create missing internal topics with the desired property values. Note that these configuration properties have practical default values.
config.storage.replication.factor
offset.storage.replication.factor
offset.storage.partitions
status.storage.replication.factor
status.storage.partitions
注釈
Confluent Platform バージョン 6.0 以降の Kafka Connect では、Kafka ブローカーの default.replication.factor
値と num.partitions
値を使用して、内部トピックを作成できるようになりました。詳細については、「Kafka ブローカーのデフォルトトピック設定の使用」を参照してください。
As each distributed worker starts up, it uses the internal Kafka topics if they already exist. If not, the worker tries to create the topics using the worker configuration properties. This gives you the option of manually creating the topics before starting Kafka Connect, if you require topic-specific settings or when Kafka Connect does not have the necessary privileges to create the topics. If you do create the topics manually, make sure to follow the guidelines provided in the list of configuration properties.
If you want to create a distributed worker that is independent of an existing Connect cluster, you must create new worker configuration properties. The following configuration properties must be different from the worker configurations used in an existing cluster:
group.id
config.storage.topic
offset.storage.topic
status.storage.topic
It is important to note that Connect clusters cannot share group IDs or
internal topics. Simply changing a group.id
will not create a new worker
separate from an existing Connect cluster. The new group.id
must also
have unique internal topics associated with it. This requires setting unique
config.storage.topic
, offset.storage.topic
, and status.storage.topic
configuration properties for the new group.id
.
You also must use different connector names than those used in the existing Connect cluster since a consumer group is created based on the connector name. Each connector in a Connect cluster shares the same consumer group.
Kafka ブローカーのデフォルトトピック設定の使用¶
Connect ワーカーでは、レプリケーション係数とパーティションの数に Kafka ブローカーのデフォルトを使用して内部トピックを作成することができます。Kafka ブローカーのデフォルトをレプリケーション係数とパーティションの数に使用するには、内部トピックのワーカーの構成プロパティに -1
を使用します。スニペットの例を以下に示します。
# Use the broker default properties
config.storage.replication.factor=-1
offset.storage.replication.factor=-1
status.storage.replication.factor=-1
offset.storage.partitions=-1
status.storage.partitions=-1
The config.storage
internal topic must always have exactly one partition.
This is why there is no option to use the Kafka broker default for
config.storage
.
トピック固有のプロパティのオーバーライド¶
内部トピックが作成される Kafka ブローカーのバージョンに対して有効な トピックプロパティ をオーバーライドすることができます。以下の例は、ワーカーの構成でこれらのプロパティを入力する方法を示しています。
config.storage.<topic-specific-property>
offset.storage.<topic-specific-property>
status.storage.<topic-specific-property>
以下の例では、レプリケーション係数に Kafka ブローカーのデフォルトのプロパティを使用し、デフォルトの同期レプリカの最小数(min.insync.replicas
)プロパティをオーバーライドしています(1
が Kafka ブローカーのデフォルトです)。
# Use the broker default properties
config.storage.replication.factor=-1
offset.storage.replication.factor=-1
status.storage.replication.factor=-1
# Override the broker default properties
config.storage.min.insync.replicas=3
offset.storage.min.insync.replicas=3
status.storage.min.insync.replicas=3
Kafka ブローカーのバージョンでトピックプロパティが有効でない場合、Connect ワーカーは起動時に失敗します。
内部トピックの手動作成¶
Kafka Connect でこれらの内部トピックの自動作成を許可することをお勧めします。ただし、手動でトピックを作成することもできます。これらのトピックを手動で作成する場合として、2 つの例を以下に示します。
- セキュリティ上の理由で、Connect などのクライアントによる Kafka トピックの作成を許可しないようにブローカーが構成されている場合があります。
- Connect によって自動では設定されない、または自動的に作成される設定とは異なる、他の高度なトピック固有の設定が必要になる場合があります。
以下のコマンド例は、Connect を起動する前に、圧縮およびレプリケートされる Kafka トピックを手動で作成する方法を示しています。パラメーターを入力する際には、 分散ワーカー のガイドラインに従ってください。
# config.storage.topic=connect-configs
bin/kafka-topics --create --bootstrap-server localhost:9092 --topic connect-configs --replication-factor 3 --partitions 1 --config cleanup.policy=compact
# offset.storage.topic=connect-offsets
bin/kafka-topics --create --bootstrap-server localhost:9092 --topic connect-offsets --replication-factor 3 --partitions 50 --config cleanup.policy=compact
# status.storage.topic=connect-status
bin/kafka-topics --create --bootstrap-server localhost:9092 --topic connect-status --replication-factor 3 --partitions 10 --config cleanup.policy=compact
All workers in a Connect cluster use the same internal topics. Workers in a different cluster must use different internal topics. For more details, see Distributed Worker Configuration.
Worker configuration properties file¶
使用するモードに関係なく、Kafka Connect ワーカーは、ワーカーの構成プロパティファイルを最初のパラメーターとして渡すことによって構成されます。以下に例を示します。
bin/connect-distributed worker.properties
Confluent Platform には、作業を開始するうえで役立つワーカーの構成プロパティファイルのサンプルが含まれています。Avro のサンプルファイルの場所は、次のとおりです。
etc/schema-registry/connect-avro-distributed.properties
etc/schema-registry/connect-avro-standalone.properties
これらのファイルのいずれかを開始点として使用します。これらのファイルには、Schema Registry と統合された Avro コンバーターを使用するために必要な構成プロパティが含まれています。これらは、ローカルで実行されている Kafka サービスおよび Schema Registry サービスと適切に連動するように構成されています。複数のブローカーを実行する必要がないため、Kafka Connect をローカルで簡単にテストすることができます。
サンプルの構成ファイルを本稼働環境のデプロイ用に編集することもできます。それには、Kafka および Schema Registry 用の正しいホスト名を使用し、内部トピックのレプリケーション係数に許容可能な値(またはデフォルト値)を使用します。
ワーカーの構成プロパティのリストについては、「Kafka Connect ワーカーの構成プロパティ」を参照してください。
キーコンバーターと値コンバーターの構成¶
The following list shows the converters packaged with the Confluent Platform:
- AvroConverter
io.confluent.connect.avro.AvroConverter
: Schema Registry とともに使用 - ProtobufConverter
io.confluent.connect.protobuf.ProtobufConverter
: Schema Registry とともに使用 - JsonSchemaConverter
io.confluent.connect.json.JsonSchemaConverter
: Schema Registry とともに使用 - JsonConverter
org.apache.kafka.connect.json.JsonConverter
(Schema Registry なし): 構造化データに使用 - StringConverter
org.apache.kafka.connect.storage.StringConverter
: シンプルな文字列フォーマット - ByteArrayConverter
org.apache.kafka.connect.converters.ByteArrayConverter
: 変換を行わない "パススルー" のオプションを提供
Kafka Connect 101 コース では、コンバーターについて詳しく説明しています。
key.converter
プロパティと value.converter
プロパティには、使用する コンバーター のタイプを指定します。すべてのコネクターのデフォルトコンバーターは、 ワーカーの構成 で指定します。ただし、どのコネクターでも、キー、値、およびヘッダーのコンバーターを完全に定義することで、デフォルトのコンバーターをオーバーライドすることができます。ほとんどのコネクターで使用できるデフォルトのキー、値、およびヘッダーのコンバーターをワーカーで定義し、別のコンバーターが必要なコネクターがある場合には、そのコネクターの構成を定義することをお勧めします。ワーカーで定義されているデフォルトの header.converter
は、StringConverter
を使用してヘッダーの値を文字列としてシリアル化し、ヘッダーの値を最も適切な数値、ブール値、配列、またはマップ表現に逆シリアル化します。スキーマはシリアル化されませんが、可能な場合は逆シリアル化するときに推測されます。
It is important to note that converter configuration properties in the worker configuration are used by all connectors running on the worker, unless a converter is added to a connector configuration.
コネクターの構成にコンバーターを追加した場合、ワーカーの構成にあるコンバータータイプのプレフィックス(key.converter.*
や value.converter.*
)が付加されたコンバータープロパティはすべて使用されません。コネクターの構成にコンバーターを追加する場合は注意してください。たとえば、ワーカーの構成に以下のような値コンバータープロパティが存在するとします。
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081
And, you add the following properties to your connector configuration:
{
"value.converter": "AvroConverter",
"value.converter.basic.auth.credentials.source": "USER_INFO",
"value.converter.basic.auth.user.info": "<username>:<password>"
}
An error will occur when the connector is started because the required Schema Registry URL
property value.converter.schema.registry.url=http://localhost:8081
is not
provided to the converter.
以下のセクションでは、コンバーターの説明と例を示します。これらのコンバーターの Schema Registry での動作については、「Kafka Connect と Schema Registry の使用」を参照してください。
Avro¶
To use the AvroConverter
with Schema
Registry, you specify the key.converter
and
value.converter
properties in the worker configuration. You must also add a
converter property that provides the Schema Registry URL. The following example shows the
AvroConverter
key and value properties that are added to the configuration:
key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://localhost:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081
The Avro key and value converters can be used independently from each other. For
example, you may want to use a StringConverter
for keys and the
AvroConverter
or JsonConverter
for values. Independent key and value
properties are shown in the following example:
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081
JSON スキーマと Protobuf¶
JSON スキーマのコンバーターと Protobuf のコンバーターはどちらも、Avro コンバーターと同じように実装されています。以下に、値コンバーターに ProtobufConverter
または JsonSchemaConverter
、キーコンバーターに StringConverter
を使用した構成例をいくつか示します。
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=io.confluent.connect.protobuf.ProtobufConverter
value.converter.schema.registry.url=http://localhost:8081
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=io.confluent.connect.json.JsonSchemaConverter
value.converter.schema.registry.url=http://localhost:8081
Avro と JSON スキーマのどちらもスキーマを JSON で表現しており、認識されていないプロパティが検出された場合にも寛容です。そのため、コンバーターはカスタム JSON プロパティを使用して、Avro および JSON スキーマには相当するものがない Kafka Connect スキーマオブジェクトをキャプチャすることができます。
一方、Protobuf には JSON と異なる独自のインターフェイス定義言語(IDL)があり、カスタムのアドホックなプロパティは許容されません。そのため、Protobuf には直接相当するものがない場合に Kafka Connect スキーマから Protobuf に変換すると、データの損失や不整合が生じる可能性があります。
たとえば、Kafka Connect のスキーマでは、int8、int16、および int32 のデータ型がサポートされています。Protobuf では、int32 および int64 がサポートされています。Connect のデータが Protobuf に変換されると、int8 と int16 のフィールドは int32 または int64 に対応付けられ、ソースが int8 または int16 であったことは示されません。
JSON スキーマでは、数値型と整数型のフィールドのみがサポートされています。ただし、JSON スキーマコンバーター( JsonSchemaConverter
)は、JSON スキーマにはそれに相当するものがないデータを connect.type
という名前のプロパティに保管します。このプロパティは JSON スキーマパーサーでは無視されるため、フィールドはダウンストリームのコンポーネントによって適切な型に復元されます。
For full encoding details, see JSON encoding for Avro and JSON
encoding for Protobuf.
Additionally, JSON Schema supports three means of combining schemas: allOf,
anyOf, and oneOf.
However, the JSON Schema converter only supports oneOf
, treating it
similarly to how the Avro converter handles unions and how the Protobuf
converter handles oneOf
.
Note that if you're configuring Avro, Protobuf, or JSON Schema converters in an environment configured for Role-Based Access Control (RBAC), see key and value converters with RBAC. For details about how converters work with Schema Registry, see Using Kafka Connect with Schema Registry.
Schema Registry では、以下のコンバーターは使用されません。
JSON(Schema Registry を使用しない)¶
Connect データに対して Schema Registry を使わずに JSON を使用する必要がある場合は、Kafka でサポートされている JsonConverter
を使用することができます。以下の例は、構成に追加される JsonConverter
のキープロパティと値プロパティを示しています。
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false
プロパティ key.converter.schemas.enable
および value.converter.schemas.enable
が true
に設定されている場合、キーまたは値はプレーンな JSON として扱われず、内部スキーマとデータの両方を含む複合 JSON オブジェクトとして扱われます。それらをソースコネクターで有効にすると、スキーマとデータの両方が複合 JSON オブジェクトに含められます。それらをシンクコネクターで有効にすると、スキーマとデータが複合 JSON オブジェクトから抽出されます。この実装では、Schema Registry が使用されないことに注意してください。
プロパティ key.converter.schemas.enable
および value.converter.schemas.enable
が false
に設定されている場合(デフォルト)、データだけが渡され、スキーマは渡されません。これにより、スキーマを必要としないアプリケーションのペイロードのオーバーヘッドが減少します。
文字列フォーマットと未加工バイト¶
org.apache.kafka.connect.storage.StringConverter
は、内部 Connect フォーマットをシンプルな文字列フォーマットに変換するために使用されます。Connect のデータをバイトに変換する場合、スキーマは無視され、データはシンプルな文字列に変換されます。バイトを Connect データフォーマットに変換する場合、コンバーターはオプションの文字列スキーマと文字列(または null)を返します。
org.apache.kafka.connect.converters.ByteArrayConverter
では、データが変換されません。バイトは変換されずにそのままコネクターを通過します。
コンバーターの詳細については、「コンバーターとシリアル化について」を参照してください。
Connect プロデューサーおよびコンシューマー¶
Kafka Connect では、内部的に Java 標準のプロデューサーとコンシューマーを使用して Kafka と通信します。Connect によって、これらのプロデューサーおよびコンシューマーのインスタンスのデフォルト設定が構成されます。これらの設定には、データが失われることなく順番に Kafka に送信されることを保証するプロパティが含まれています。
Default Connect producer properties¶
By default, Connect configures the Kafka producers for source connectors with the following properties:
- プロデューサーのブートストラップサーバーで、Connect クラスターで使用されているのと同じ Kafka クラスターを参照します。
- コネクターのキーおよび値のコンバーターで動作するキーおよび値のシリアライザーを構成します。
- パターン
connector-producer-<connectorName>-<taskId>
を使用して、コネクターとタスクに基づいてプロデューサーのclient.id
を生成します。 acks=all
を設定して、生成されたメッセージがすべての同期レプリカ(ISR)に適切に書き込まれるようにします。- 再試行可能な例外については、無限の再試行でデータが重複する可能性を低減するため、Connect により以下のプロパティでプロデューサーが構成されます。
request.timeout.ms=<max>
max.block.ms=<max>
max.in.flight.requests.per.connection=1
delivery.timeout.ms=<max>
You can override the defaults by using the producer.*
properties in the
worker configuration, or by using the producer.override.*
properties in
connector configurations, but changing these default properties may compromise
the delivery guarantees of Connect.
Producer and consumer overrides¶
前のセクションで説明した以外のデフォルト設定をオーバーライドする必要が生じる場合があります。そのような必要が生じる場合の例を 2 つ、以下に示します。
ワーカーのオーバーライドの例
ログファイルのコネクターを実行するスタンドアロンプロセスを考えてみましょう。ログ収集では、低レイテンシでベストエフォート型のデリバリーが適している場合があります。これは、クライアントでのデータのバッファリングを避けるため、接続の問題があっても、小規模なデータの損失をアプリケーションが許容できる場合です。これにより、ログ収集を可能な限り軽量化しておくことができます。
To override producer configuration
properties and
consumer configuration
properties for all connectors
controlled by the worker, you prefix worker configuration properties with
producer.
or consumer.
as shown in the following example:
producer.retries=1
consumer.max.partition.fetch.bytes=10485760
The previous example overrides the default producer retries
property to
retry sending messages only one time. The consumer override increases the
default amount of data fetched from a partition per request to 10 MB. These
configuration changes are applied to all connectors controlled by the worker. Be
careful making any changes to these settings when running distributed mode
workers.
コネクターごとのオーバーライドの例
By default, the producers and consumers used for connectors are created using the same properties that Connect uses for its own internal topics. This means that the same Kafka principal must be able to read and write to all the internal topics and all of the topics used by the connectors.
You may want the producers and consumers used for connectors to use a different
Kafka principal. It is possible for connector configurations to override worker
properties used to create producers and consumers. These are prefixed with
producer.override.
and consumer.override.
. For more information about
per-connector overrides, see Override the Worker
Configuration.
プロデューサーとコンシューマーの詳細については、「Kafka プロデューサー」と「Kafka コンシューマー」を参照してください。構成プロパティのリストについては、「プロデューサーの構成プロパティ」と「コンシューマーの構成プロパティ」を参照してください。
ソースコネクターのトピックの自動作成の構成¶
Beginning with Confluent Platform version 6.0, Kafka Connect can create topics for source
connectors if the topics do not exist on the Apache Kafka® broker. To use auto topic
creation for source connectors, you must set the Connect worker property to
true
for all workers in the Connect cluster. In addition, you must
create supporting properties in each source connector configuration.
重要
- This feature does not affect sink connectors or their configuration. Any topic creation properties you add to sink connectors will be ignored and will produce a warning in the log.
- If you do not want the source connector to create topics that are missing,
you must disabled the feature by setting
topic.creation.enable=false
in the Connect worker.
ワーカープロパティ¶
ソースコネクターのトピックの自動作成を有効または無効にするには、次のワーカープロパティを使用します。
topic.creation.enable
Defaults to
true
. This feature is enabled only for source connector configurations that have the supporting ソースコネクターのプロパティ.- 型: boolean
- デフォルト: true
- 重要度: 低
ソースコネクターのプロパティ¶
Several source connector properties are associated with the worker property
topic.creation.enable
. These properties set the default replication factor,
number of partitions, and other topic-specific settings to be used by
Kafka Connect to create a topic if it does not exist. None of the properties
have default values.
The following source connector configuration properties are required:
topic.creation.$alias.replication.factor
topic.creation.$alias.partitions
The auto topic creation feature is enabled for the source connector only when
the feature is enabled in the worker configuration and when the source connector
configuration specifies the required replication factor and number of
partitions for one group. Users may choose to use the default values specified
in the Kafka broker by setting topic.creation.$alias.replication.factor
or
topic.creation.$alias.partitions
to -1
.
You can define more connector properties using configuration property groups.
Configuration property groups are added using the property
topic.creation.groups
. The hierarchy of groups is built on top of a single
foundational group called the default
configuration property group. The
default
group always exists and does not need to be listed in the
topic.creation.groups
property in the connector configuration. Including
default
in topic.creation.groups
results in a warning.
The following source connector configuration properties are used in association
with the topic.creation.enable=true
worker property. For example properties,
see 構成の例.
注釈
構成プロパティには、Java regex として定義されている正規表現(regex)を使用できます。
topic.creation.groups
一致するトピックにグループ別のトピック構成を定義するために使用するグループ別名のリスト
default
グループは常に存在し、すべてのトピックに一致します。- 型: String 型のリスト
- デフォルト: 空
- 指定可能な値: このプロパティの値には、任意の追加グループを指定できます。トピックの構成には常に
default
グループが定義されています。
topic.creation.$alias.replication.factor
コネクターで作成する新規トピックのレプリケーション係数。この値は、Kafka クラスターのブローカーの数を超えてはなりません。この値が Kafka ブローカーの数よりも大きい場合、コネクターがトピックの作成を試行するとエラーが発生します。
default
グループの場合、これは 必須のプロパティ です。topic.creation.groups
で定義されている他のグループの場合、このプロパティは省略可能です。他のグループでは、Kafka ブローカーのデフォルト値が使用されます。- 型: int
- デフォルト: なし
- 指定可能な値 : 具体的な有効値を指定する場合は
>= 1
で指定し、Kafka ブローカーのデフォルト値を使用する場合は-1
を指定します。
topic.creation.$alias.partitions
このコネクターによって作成されたトピックのパーティションの数。
default
グループの場合、これは 必須のプロパティ です。topic.creation.groups
で定義されている他のグループの場合、このプロパティは省略可能です。他のグループでは、Kafka ブローカーのデフォルト値が使用されます。- 型: int
- デフォルト: なし
- 指定可能な値 : 具体的な有効値を指定する場合は
>= 1
で指定し、Kafka ブローカーのデフォルト値を使用する場合は-1
を指定します。
topic.creation.$alias.include
トピック名に一致する正規表現を表す文字列のリスト。このリストは、一致する値を持つトピックを対象に含め、このグループの特定の構成を一致するトピックに適用するために使用します。
topic.creation.groups
で定義されているすべてのグループに、$alias
が適用されます。default
グループには、このプロパティは適用されません。- 型: String 型のリスト
- デフォルト: 空
- 指定可能な値: 正確なトピック名または正規表現のコンマ区切りのリスト。
topic.creation.$alias.exclude
トピック名に一致する正規表現を表す文字列のリスト。このリストは、一致する値を持つトピックを、グループの特定の構成の適用から除外するために使用します。
topic.creation.groups
で定義されているすべてのグループに、$alias
が適用されます。default
グループには、このプロパティは適用されません。トピックの除外ルールは、すべての包含ルールをオーバーライドすることに注意してください。- 型: String 型のリスト
- デフォルト: 空
- 指定可能な値: 正確なトピック名または正規表現のコンマ区切りのリスト。
topic.creation.$alias.${kafkaTopicSpecificConfigName}
レコードの書き込み先の Kafka ブローカーのバージョンに対するすべての ブローカー構成の動的な変更 。ルールに対して構成が指定されていない場合、ブローカーのトピックレベルの構成値が使用されます。
default
グループ、およびtopic.creation.groups
で定義されているすべてのグループに、$alias
が適用されます。- 型: プロパティ値
- デフォルト : Kafka ブローカー値
構成の例¶
The following example configuration snippets show how the source connector
configuration properties are entered when topic.creation.enable
is enabled
in the Connect worker configuration.
Example 1
All new topics created by Connect have a replication factor of 3 and 5
partitions. default
is the only group, so config topic.creation.groups
is not used.
...omitted
topic.creation.default.replication.factor=3
topic.creation.default.partitions=5
...omitted
Example 2
New topics created by Connect have replication factor of 3 and 5 partitions.
The exception to this configuration setting are topics that match the inclusion
list of the inorder
group, which have one partition.
...omitted
topic.creation.groups=inorder
topic.creation.default.replication.factor=3
topic.creation.default.partitions=5
topic.creation.inorder.include=status, orders.*
topic.creation.inorder.partitions=1
...omitted
Example 3
New topics created by Connect have replication factor of 3 and 5 partitions.
The key_value_topic
and another.compacted.topic
topics that begin with
the prefix configurations
are compacted and have a replication factor of 5
and one partition.
...omitted
topic.creation.groups=compacted
topic.creation.default.replication.factor=3
topic.creation.default.partitions=5
topic.creation.compacted.include=key_value_topic, another.compacted.topic, configurations.*
topic.creation.compacted.replication.factor=5
topic.creation.compacted.partitions=1
topic.creation.compacted.cleanup.policy=compact
...omitted
Example 4
New topics created by Connect have replication factor of 3 and 5 partitions.
Topics that begin with the prefix configurations
are compacted. Topics that
match the includes list of highly_parallel
and don't match its exclusion
list have a replication factor of 1 and one partition.
...omitted
topic.creation.groups=compacted, highly_parallel
topic.creation.default.replication.factor=3
topic.creation.default.partitions=5
topic.creation.highly_parallel.include=hpc.*,parallel.*
topic.creation.highly_parallel.exclude=.*internal, .*metadata, .*config.*
topic.creation.highly_parallel.replication.factor=1
topic.creation.highly_parallel.partitions=100
topic.creation.compacted.include=configurations.*
topic.creation.compacted.cleanup.policy=compact
...omitted
セキュリティ構成の例¶
When security is configured, the Connect worker should have a principal
configured to DESCRIBE and CREATE topics, which is inherited by all connectors
in the worker. If different security
settings are required from what the
Connect worker configuration provides, you can add producer.overrride
properties to the source connector configuration to provide the security
credentials, as shown in the following examples:
...omitted
producer.override.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username="alice" \
password="alice-secret";
...omitted
...omitted
producer.override.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username="alice" \
password="alice-secret";
admin.override.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username="bob" \
password="bob-secret";
...omitted
An error is logged and the task fails if topic.creation.enable=true
is
configured, and neither the worker properties nor the connector overrides allow
for creation of new topics.
Connect Reporter¶
Kafka Connect Reporter は、シンク操作の結果をレポータートピックに送信します。レコードのシンクに成功した後、またはエラー条件に従って、結果レポートを提出するために Connect Reporter が呼び出されます。このレポートは、シンクイベントに関する追加情報とともに、元のレコードの処理方法に関する詳細が含められるように構成されています。それらのレコードは、消費できるように、構成可能な成功トピックおよびエラートピックに書き込まれます。シンクコネクター構成に追加されている基本的な Connect Reporter 構成プロパティの例を以下に示します。
reporter.bootstrap.servers=localhost:9092
reporter.result.topic.name=success-responses
reporter.result.topic.replication.factor=1
reporter.error.topic.name=error-responses
reporter.error.topic.replication.factor=1
Connect Reporter を完全に無効にするには、「 Connect Reporter の無効化」を参照してください。
環境がセキュアな場合は、管理クライアントとプロデューサーの両方に対して構成ブロックを使用します。プロデューサーは、レポータートピックにレコードを送信するように構築されています。管理クライアントは、トピックを作成します。認証情報は、セキュアな環境で追加する必要があります。管理プロパティとプロデューサープロパティの例を以下に示します。
reporter.admin.bootstrap.servers=localhost:9092
reporter.admin.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule \
required username="<username>" password="<password>";
reporter.admin.security.protocol=SASL_SSL
reporter.admin.sasl.mechanism=PLAIN
reporter.producer.bootstrap.servers=localhost:9092
reporter.producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule \
required username="<username>" password="<password>";
reporter.producer.security.protocol=SASL_SSL
reporter.producer.sasl.mechanism=PLAIN
Additional Reporter configuration property examples are provided in each applicable Kafka Connect sink connector document. For an example, see the Reporter properties in the HTTP Sink connecter.
Reporter と Kerberos のセキュリティ¶
The following configuration example shows a sink connector with all the necessary configuration properties for Reporter and Kerberos security. This example shows the Prometheus Metrics Sink Connector for Confluent Platform, but can be modified for any applicable sink connector.
{
"name" : "prometheus-connector",
"config" : {
"topics":"prediction-metrics",
"connector.class" : "io.confluent.connect.prometheus.PrometheusMetricsSinkConnector",
"tasks.max" : "1",
"confluent.topic.bootstrap.servers":"localhost:9092",
"confluent.topic.ssl.truststore.location":"/etc/pki/hadoop/kafkalab.jks",
"confluent.topic.ssl.truststore.password":"xxxx",
"confluent.topic.ssl.keystore.location":"/etc/pki/hadoop/kafkalab.jks",
"confluent.topic.ssl.keystore.password":"xxxx",
"confluent.topic.ssl.key.password":"xxxx",
"confluent.topic.security.protocol":"SASL_SSL",
"confluent.topic.replication.factor": "3",
"confluent.topic.sasl.kerberos.service.name":"kafka",
"confluent.topic.sasl.jaas.config":"com.sun.security.auth.module.Krb5LoginModule required \nuseKeyTab=true \nstoreKey=true \nkeyTab=\"/etc/security/keytabs/svc.kfkconnect.lab.keytab\" \nprincipal=\"svc.kfkconnect.lab@DS.DTVENG.NET\";",
"prometheus.scrape.url": "http://localhost:8889/metrics",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false",
"behavior.on.error": "LOG",
"reporter.result.topic.replication.factor": "3",
"reporter.error.topic.replication.factor": "3",
"reporter.bootstrap.servers":"localhost:9092",
"reporter.producer.ssl.truststore.location":"/etc/pki/hadoop/kafkalab.jks",
"reporter.producer.ssl.truststore.password":"xxxx",
"reporter.producer.ssl.keystore.location":"/etc/pki/hadoop/kafkalab.jks",
"reporter.producer.ssl.keystore.password":"xxxx",
"reporter.producer.ssl.key.password":"xxxx",
"reporter.producer.security.protocol":"SASL_SSL",
"reporter.producer.sasl.kerberos.service.name":"kafka",
"reporter.producer.sasl.jaas.config":"com.sun.security.auth.module.Krb5LoginModule required \nuseKeyTab=true \nstoreKey=true \nkeyTab=\"/etc/security/keytabs/svc.kfkconnect.lab.keytab\" \nprincipal=\"svc.kfkconnect.lab@DS.DTVENG.NET\";",
"reporter.admin.ssl.truststore.location":"/etc/pki/hadoop/kafkalab.jks",
"reporter.admin.ssl.truststore.password":"xxxx",
"reporter.admin.ssl.keystore.location":"/etc/pki/hadoop/kafkalab.jks",
"reporter.admin.ssl.keystore.password":"xxxx",
"reporter.admin.ssl.key.password":"xxxx",
"reporter.admin.security.protocol":"SASL_SSL",
"reporter.admin.sasl.kerberos.service.name":"kafka",
"reporter.admin.sasl.jaas.config":"com.sun.security.auth.module.Krb5LoginModule required \nuseKeyTab=true \nstoreKey=true \nkeyTab=\"/etc/security/keytabs/svc.kfkconnect.lab.keytab\" \nprincipal=\"svc.kfkconnect.lab@DS.DTVENG.NET\";",
"confluent.license":"eyJ0eXAiOiJK ...omitted"
}
Connect Reporter の無効化¶
Connect Reporter を無効にするには、reporter.error.topic.name
構成プロパティと reporter.result.topic.name
構成プロパティに空の文字列を設定します。
注釈
reporter.bootstrap.servers
プロパティは必須ではなく、デフォルト値は空です。しかし、これらのトピック名の構成プロパティは、デフォルトで空ではないため、reporter.bootstrap.servers
が必須になります。
reporter.error.topic.name = ${connector}-error
reporter.result.topic.name = ${connector}-success
Metrics reporter metadata¶
Confluent Platform 6.0 のリリースに伴い、Kafka Connect では、コネクターがメトリクスのコンテキストメタデータを組み込みのメトリクスクライアントライブラリに渡すための構成プレフィックスが利用可能になりました。Connect ワーカーは、インスタンス化されたすべてのコネクターにコンテキストの構成を渡します。これらの ワーカーの構成プロパティ に関する情報を以下に示します。
metrics.context.<key>=<value>
構成されたキー/値ペアは、Connect ワーカー内のコネクターによって、MetricsContext
インターフェイスを使用して、構成された Confluent Metrics Reporter に渡されます。たとえば、metrics.context.foo.bar=baz
と設定して Connect ワーカーを構成すると、MetricsContext
メタデータの Metrics Reporter の値 baz
にマップされたフィールド foo.bar
が追加されます。
Connect ワーカーは以下のフィールドを MetricsContext
に渡すことができます。
connect.kafka.cluster.id
。Apache Kafka® をサポートしているクラスターを示します。connect.group.id
to indicate the group ID used for Connect worker coordination.connect.group.id
is only valid for distributed mode. For more information, see Connect の内部トピック.
ConfigProvider Interface¶
ConfigProvider クラスのインターフェイスでは、起動時に動的に解決されるワーカー構成の変数を使用することができます。また、コネクターが(再)起動されたときに動的に解決されるコネクター構成の変数を使用することもできます。構成プロパティ値の中の変数をシークレットの代わりに使用したり、実行時に動的に解決する必要がある情報の代わりに使用したりすることができます。
注釈
コネクターの構成は、Connect REST API の変数で保持され、共有されます。コネクターの起動時にのみ、メモリー内の変数の一時的な解決や置き換えが行われます。シークレットは、コネクター構成、ログ、または REST API のリクエストや応答の中には 保持されません。
Connect ワーカーは、変数の解決に、ワーカー構成で定義されている名前付きの ConfigProviders に依存します。各変数では、使用するべき ConfigProvider の名前と、ConfigProvider が変数を置換文字列に解決するために使用する情報を指定します。
ConfigProvider
の実装はすべて、標準的な Java の ServiceLoader
メカニズムを使用して検出されます。ConfigProvider
のカスタム実装を作成するには、ConfigProvider インターフェイスを実装します。実装クラス、および ConfigProvider
実装クラスの完全修飾名を含めた META-INF/services/org.apache.kafka.common.config.provider.ConfigProvider
という名前のファイルを JAR ファイルにパッケージ化します。JAR ファイルでは、Connect フレームワークで提供されているもの以外のサードパーティのライブラリを使用できますが、以下に示すように JAR ファイルと一緒にインストールする必要があることに注意してください。
ConfigProvider
のカスタム実装をインストールするには、JAR ファイルを含む新しいサブディレクトリを Connect の plugin.path
にあるディレクトリに追加し、Connect ワーカーを(再)起動します。Connect ワーカーが起動すると、ワーカー構成で指定された ConfigProvider
のすべての実装がインスタンス化されます。config.providers.[provider].param.
のプレフィックスが付加されているプロパティはすべて、ConfigProvider
の configure()
メソッドに渡されます。Connect ワーカーがシャットダウンされると、ConfigProvider
の close()
メソッドが呼び出されます。
重要
Connect クラスター内のすべてのワーカーは、ワーカー構成内のすべての変数と、すべてのコネクター構成で使用されるすべての変数を解決できなければなりません。そのためには、以下の要件を満たしている必要があります。
- Connect クラスター内のすべてのワーカーに、同じ一連の名前付き構成プロバイダーが必要です。
- 各ワーカーの各プロバイダーには、ワーカー構成またはコネクター構成で使用される変数を解決するために必要なリソースへのアクセス権限が付与されている必要があります。
これを実現するため、分散ワーカーの構成には以下の構成プロパティが追加されています。
config.providers
: プロバイダーの名前のコンマ区切りリスト。config.providers.{name}.class
: プロバイダーの Java クラス名。config.providers.{name}.param.{param-name}
: 初期化時に上記の Java クラスに渡すパラメーター。
FileConfigProvider¶
Kafka は、変数の参照を各ワーカーのローカルファイルの値に置き換えることができる、FileConfigProvider
と呼ばれる ConfigProvider
の実装を提供しています。たとえば、構成プロパティにシークレットを設定するのではなく、ローカルファイルにシークレットを保存して、コネクター構成の変数を使用することができます。コネクターが起動されると、Connect は、ファイル構成プロバイダーを使用して変数を解決し、実際のシークレットに置き換えます。このようにすることで、コネクターの構成が Connect REST API で保持され、共有される場合に、コネクターの構成にシークレットを含めないようにすることができます。
重要
Connect クラスター内の すべてのワーカー は、構成プロバイダーを参照するすべての変数によって参照されるファイルにアクセスできる必要があります。
FileConfigProvider
を名前で参照する変数は、${provider:[path:]key}
という形式を取る必要があります。path は、各 Connect ワーカーのプロパティファイルの完全修飾パス、key はそのプロパティファイル内のキーの名前です。FileConfigProvider
では、各変数にパス(およびそのファイルのプロパティキー)が指定されているすべてのファイルの読み取りがサポートされています。Connect は、このような変数を解決するとき、プロパティファイルを読み取り、対応するキーの値を抽出して、変数全体をその値に置き換えます。
データベースの URL、ユーザー名、パスワードを含む JDBC コネクター構成を以下に示します。
connection.url=jdbc:oracle:thin:@myhost:1521:orcl
connection.user=scott
connection.password=<my-secret-password>
これらの詳細をコネクター構成で公開する代わりに、FileConfigProvider
を使用して、Connect の各ワーカーがアクセスできるファイルに保管することで、他の OS ユーザーから保護することができます。以下の例では、/opt/connect-secrets.properties
という名前の別のファイルを示します。/opt/connect-secrets.properties
に追加されているプロパティは次のとおりです。
productsdb-url=jdbc:oracle:thin:@myhost:1521:orcl
productsdb-username=scott
productsdb-password=my-secret-password
other-connector-url=jdbc:oracle:thin:@myhost:1521:orcl
other-connector-username=customers
other-connector-password=superSecret!
各 Connect ワーカーが FileConfigProvider
を使用するように構成することができます。ワーカーの構成には、以下のプロパティが含まれます。
# Additional properties added to the worker configuration
config.providers=file
config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProvider
JDBC コネクターの構成で、シークレットの代わりに "変数" を使用できるようになりました。
# Additional properties added to the connector configuration
connection.url=${file:/opt/connect-secrets.properties:productsdb-url}
connection.user=${file:/opt/connect-secrets.properties:productsdb-username}
connection.password=${file:/opt/connect-secrets.properties:productsdb-password}
別のコネクターの構成では、別のファイルの変数を使用することや、同じファイル内の別のプロパティを使用する変数を使用することができます。
# Additional properties added to another connector configuration
connection.url=${file:/opt/connect-secrets.properties:other-connector-url}
connection.user=${file:/opt/connect-secrets.properties:other-connector-username}
connection.password=${file:/opt/connect-secrets.properties:other-connector-password}
InternalSecretConfigProvider¶
Confluent Platform は、Connect の シークレットレジストリ で使用される InternalSecretConfigProvider
という名前の別の ConfigProvider
の実装を提供しています。シークレットレジストリはシークレットを提供するレイヤーであり、これにより、Connect が暗号化された Connect 認証情報を、REST API で公開されるトピックに保管できるようにします。これにより、実際のコネクターの構成に、暗号化されていない認証情報が含まれないようにすることができます。以下の例は、ワーカーの構成ファイルで InternalSecretConfigProvider
を構成する方法を示しています。
### Secret Provider
config.providers=secret
config.providers.secret.class=io.confluent.connect.secretregistry.rbac.config.provider.InternalSecretConfigProvider
config.providers.secret.param.master.encryption.key=<encryption key>
config.providers.secret.param.kafkastore.bootstrap.servers=SASL_PLAINTEXT://<Kafka broker URLs>
config.providers.secret.param.kafkastore.security.protocol=SASL_PLAINTEXT
config.providers.secret.param.kafkastore.sasl.mechanism=OAUTHBEARER
config.providers.secret.param.kafkastore.sasl.login.callback.handler.class=io.confluent.kafka.clients.plugins.auth.token.TokenUserLoginCallbackHandler
config.providers.secret.param.kafkastore.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
username="<service-principal-username>" \
password="<service-principal-password>" \
metadataServerUrls="<metadata server URLs>";
Shutting Down Kafka Connect¶
There are a few ways to shut down Kafka Connect. If you are running Connect in Confluent Platform standalone mode, you can use the following local Confluent CLI command:
confluent local services connect stop
詳細については、「confluent local services connect stop」を参照してください。
If you're not using the Confluent CLI, or if you're running Connect in distributed mode, perform following steps to shut down Kafka Connect:
Connect ノードのいずれかでターミナルセッションを開始します。
Search for the running Connect process:
ps aux | grep ConnectDistributed
出力で PID を特定します。
<node-hostname> <worker-pid> 0.2 2.1 8414400 351944 s003 S 12:42PM 2:52.62 /path/to/your/bin/java ...omitted... org.apache.kafka.connect.cli.ConnectDistributed /path/to/your/connect.properties
Stop the process:
kill <worker-pid>
残りのすべての Connect ノードで、この実行中のプロセスを停止します。
重要
Be sure not to use kill -9
to stop the process.
次のステップ¶
After getting started with your deployment, you may want check out the following Kafka Connect documentation:
- Kafka Connect 101 トレーニングコース
- Building data pipelines with Apache Kafka トレーニングコース
- チュートリアル : Kafka との間のデータの移動
- Kafka Connect のログ記録
- Kafka Connect のアップグレード
- Kafka Connect のセキュリティ
- Kafka Connect REST インターフェイス
- Kafka Connect と Schema Registry の使用
- コネクタープラグインのアップグレード
- ワーカーの構成のオーバーライド
- Adding Connectors or Software (Docker)
ちなみに
Try out Confluent's end-to-end demos for Kafka Connect on-premises, Confluent Cloud, and Confluent for Kubernetes.