トピックの構成¶

このトピックでは、Confluent Platform で使用できる構成パラメーターについて説明します。パラメーターは、重要度の高いものから順に並べられています。

cleanup.policy

"delete" または "compact" あるいは両方の文字列。この文字列は、古いログセグメントに使用する保持ポリシーを指定します。デフォルトのポリシー（"delete"）では、保持時間またはサイズ制限に達したときに古いセグメントを破棄します。"compact" 設定は、トピックでのログ圧縮を有効にします。

型: リスト

デフォルト: delete

指定可能な値: [compact、delete]

サーバーのデフォルトプロパティ: log.cleanup.policy

重要度: 中

compression.type

特定のトピックの最終的な圧縮タイプを指定します。この構成には、標準の圧縮コーデック（「gzip」、「snappy」、「lz4」、「zstd」）を指定できます。他に使用できる値として、「uncompressed」を指定すると圧縮なしと同等になり、「producer」を指定するとプロデューサーで設定された元の圧縮コーデックが維持されます。

型:	string
デフォルト:	プロデューサー
指定可能な値:	[uncompressed、zstd、lz4、snappy、gzip、producer]
サーバーのデフォルトプロパティ:	compression.type
重要度:	中

confluent.key.schema.validation

レコードキーでのスキーマ検証がこのトピックで有効になっている場合、true です。

型: boolean

デフォルト: false

指定可能な値:

サーバーのデフォルトプロパティ: null

重要度: 中
confluent.key.subject.name.strategy

キースキーマをスキーマレジストリに登録する際に使用されるサブジェクト名の構築方法を決定します。デフォルトでは TopicNameStrategy が使用されます。

型: string

デフォルト: io.confluent.kafka.serializers.subject.TopicNameStrategy

指定可能な値:

サーバーのデフォルトプロパティ: null

重要度: 中
confluent.tier.enable

トピックの階層化を有効にします。これにより、構成されたリモートストレージとの間で、データの階層化とフェッチが有効になります。

型: boolean

デフォルト: false

指定可能な値:

サーバーのデフォルトプロパティ: confluent.tier.enable

重要度: 中
confluent.tier.local.hotset.bytes

階層化が有効になっている場合、この構成は、ブローカーローカルストレージで（ログセグメントから構成される）パーティションを拡張できる最大サイズを制御します。このサイズに達すると、空き領域を増やすために古いログセグメントが破棄されます。ブローカーローカルストレージに保持されるログセグメントは、"ホットセット" と呼ばれます。ローカルストアから破棄されたセグメントは、保持の構成によっては引き続き階層型ストレージに存在し、フェッチ可能な状態のまま残る可能性があります。デフォルトではサイズの制限はなく、時間の制限があるだけです。この制限はパーティションレベルで適用されるため、トピックのホットセットのバイト数を計算するには、パーティション数を乗算する必要があります。

型: long

デフォルト: -1

指定可能な値:

サーバーのデフォルトプロパティ: confluent.tier.local.hotset.bytes

重要度: 中
confluent.tier.local.hotset.ms

階層化が有効になっている場合、この構成は、ブローカーローカルストレージにログセグメントを保持する最大時間を制御します。この時間に達すると、空き領域を増やすために古いログセグメントが破棄されます。ローカルストアから破棄されたセグメントは、保持の構成によっては引き続き階層型ストレージに存在し、フェッチ可能な状態のまま残る可能性があります。-1 に設定されている場合、時間制限は適用されません。

型: long

デフォルト: 86400000（1 日）

指定可能な値:

サーバーのデフォルトプロパティ: confluent.tier.local.hotset.ms

重要度: 中
confluent.value.schema.validation

レコード値でのスキーマ検証がこのトピックで有効になっている場合、true です。

型: boolean

デフォルト: false

指定可能な値:

サーバーのデフォルトプロパティ: null

重要度: 中
confluent.value.subject.name.strategy

値スキーマをスキーマレジストリに登録する際に使用されるサブジェクト名の構築方法を決定します。デフォルトでは TopicNameStrategy が使用されます。

型: string

デフォルト: io.confluent.kafka.serializers.subject.TopicNameStrategy

指定可能な値:

サーバーのデフォルトプロパティ: null

重要度: 中
delete.retention.ms

ログ圧縮トピックの delete tombstone マーカーを保持する時間。この設定でも、オフセット 0 から始める場合に、コンシューマーが読み取りを完了する必要がある時間を制限します。これによって、最終段階の有効なスナップショットを確実に取得できます（そうでない場合、スキャンを完了する前に delete tombstone が収集される可能性があります）。

型: long

デフォルト: 86400000（1 日）

指定可能な値: [0,...]

サーバーのデフォルトプロパティ: log.cleaner.delete.retention.ms

重要度: 中
file.delete.delay.ms

ファイルシステムからファイルを削除するまでの待機時間

型: long

デフォルト: 60000（1 分）

指定可能な値: [0,...]

サーバーのデフォルトプロパティ: log.segment.delete.delay.ms

重要度: 中
flush.messages

この設定により、ログに書き込まれたデータの fsync を強制的に実行する間隔を指定できます。たとえば、これを 1 に設定した場合、メッセージ 1 件ごとに fsync を実行します。これが 5 の場合、メッセージ 5 件ごとに fsync を実行します。一般的には、これを設定せず、永続性のためのレプリケーションを使用して、より効率的なオペレーティングシステムのバックグランドフラッシュ機能を許可することをお勧めします。この設定は、トピックごとにオーバーライドできます（トピックごとの構成に関するセクションを参照）。

型: long

デフォルト: 9223372036854775807

指定可能な値: [0,...]

サーバーのデフォルトプロパティ: log.flush.interval.messages

重要度: 中
flush.ms

この設定により、ログに書き込まれたデータの fsync を強制的に実行する時間の間隔を指定できます。たとえば、これを 1000 に設定した場合、1000 ミリ秒が経過した後で fsync を実行します。一般的には、これを設定せず、永続性のためのレプリケーションを使用して、より効率的なオペレーティングシステムのバックグランドフラッシュ機能を許可することをお勧めします。

型: long

デフォルト: 9223372036854775807

指定可能な値: [0,...]

サーバーのデフォルトプロパティ: log.flush.interval.ms

重要度: 中

follower.replication.throttled.replicas

ログレプリケーションをフォロワー側でスロットルする必要があるレプリカのリスト。このリストには、レプリカのセットを [PartitionId]:[BrokerId],[PartitionId]:[BrokerId]:... の形式で指定する必要があります。または、ワイルドカード「*」を使用して、このトピックのすべてのレプリカをスロットルすることもできます。

型:	リスト
デフォルト:	""
指定可能な値:	[partitionId]:[brokerId],[partitionId]:[brokerId],...
サーバーのデフォルトプロパティ:	follower.replication.throttled.replicas
重要度:	中

index.interval.bytes

この設定は、Kafka がインデックスエントリをオフセットインデックスに追加する頻度を制御します。デフォルト設定では、おおよそ 4096 バイトごとにメッセージのインデックスを作成します。インデックス作成をさらに頻繁にすると、ログ内で読み取りがジャンプする位置がさらに正確になりますが、インデックスのサイズが大きくなります。おそらく、これを変更する必要はないでしょう。

型: int

デフォルト: 4096（4 キビバイト）

指定可能な値: [0,...]

サーバーのデフォルトプロパティ: log.index.interval.bytes

重要度: 中

leader.replication.throttled.replicas

ログレプリケーションをリーダー側でスロットルする必要があるレプリカのリスト。このリストには、レプリカのセットを [PartitionId]:[BrokerId],[PartitionId]:[BrokerId]:... の形式で指定する必要があります。または、ワイルドカード「*」を使用して、このトピックのすべてのレプリカをスロットルすることもできます。

型:	リスト
デフォルト:	""
指定可能な値:	[partitionId]:[brokerId],[partitionId]:[brokerId],...
サーバーのデフォルトプロパティ:	leader.replication.throttled.replicas
重要度:	中

max.compaction.lag.ms

ログ内でメッセージが圧縮対象外として保持される最大時間。圧縮中のログにのみ適用されます。

型: long

デフォルト: 9223372036854775807

指定可能な値: [1,...]

サーバーのデフォルトプロパティ: log.cleaner.max.compaction.lag.ms

重要度: 中
max.message.bytes

Kafka で許容される最大レコードバッチサイズ（圧縮が有効な場合は圧縮後のサイズを表します）。0.10.2 よりも前のコンシューマーが存在する環境でこの値を大きくする場合は、それだけの大きさのレコードバッチをフェッチできるように、コンシューマーのフェッチサイズも大きくする必要があります。最新バージョンのメッセージフォーマットでは、効率を上げるためにレコードが常にバッチにグループ化されます。前のメッセージフォーマットバージョンでは非圧縮のレコードはバッチにグループ化されず、その場合、この制限は 1 つのレコードにのみ適用されます。

型: int

デフォルト: 1048588

指定可能な値: [0,...]

サーバーのデフォルトプロパティ: message.max.bytes

重要度: 中

ブローカーがメッセージをログに追加するときに使用するメッセージフォーマットバージョンを指定します。値は有効な ApiVersion であることが必要です。例として、0.8.2、0.9.0.0、0.10.0 などがあります。詳細については ApiVersion を確認してください。特定のメッセージフォーマットバージョンを設定すると、ディスク上にあるすべての既存のメッセージが、指定のバージョン以前であると保証することになります。この値の設定が正しくない場合、以前のバージョンを使用するコンシューマーは、受け取ったメッセージのフォーマットに対応していないために動作しなくなります。

型:	string
デフォルト:	2.6-IV0
指定可能な値:	[0.8.0, 0.8.1, 0.8.2, 0.9.0, 0.10.0-IV0, 0.10.0-IV1, 0.10.1-IV0, 0.10.1-IV1, 0.10.1-IV2, 0.10.2-IV0, 0.11.0-IV0, 0.11.0-IV1, 0.11.0-IV2, 1.0-IV0, 1.1-IV0, 2.0-IV0, 2.0-IV1, 2.1-IV0, 2.1-IV1, 2.1-IV2, 2.2-IV0, 2.2-IV1, 2.3-IV0, 2.3-IV1, 2.4-IV0, 2.4-IV1, 2.5-IV0, 2.6-IV0]
サーバーのデフォルトプロパティ:	log.message.format.version
重要度:	中

message.timestamp.difference.max.ms

ブローカーがメッセージを受信した時点のタイムスタンプと、メッセージ内に指定されたタイムスタンプとの間で許容される最大差異。message.timestamp.type=CreateTime の場合、タイムスタンプの相違がこのしきい値を超えると、メッセージは拒否されます。この構成は、message.timestamp.type=LogAppendTime の場合、無視されます。

型: long

デフォルト: 9223372036854775807

指定可能な値: [0,...]

サーバーのデフォルトプロパティ: log.message.timestamp.difference.max.ms

重要度: 中
message.timestamp.type

メッセージのタイムスタンプが、メッセージの作成時刻とログの追加時刻のどちらであるかを示します。使用できる値は `CreateTime` または `LogAppendTime` です。

型: string

デフォルト: CreateTime

指定可能な値: [CreateTime、LogAppendTime]

サーバーのデフォルトプロパティ: log.message.timestamp.type

重要度: 中
min.cleanable.dirty.ratio

この構成では、ログコンパクターがログのクリーニングを試行する頻度を制御します（ログ圧縮が有効になっていることが前提）。デフォルトでは、ログの 50% 以上が圧縮されている場合、ログのクリーニングは回避されます。この割合によって、重複により無駄になるログの最大スペースを制限します（50% の場合、ログの最大 50% まで重複にすることができます）。割合が高くなるほど、クリーニングの回数が減り、その効率が向上しますが、ログで無駄になるスペースは増えます。max.compaction.lag.ms 構成または min.compaction.lag.ms 構成も指定した場合、次のいずれかの状態になった時点で、ログコンパクターはそのログが圧縮の対象であるとみなします。(i) ダーティの比率のしきい値に達し、ログがダーティな（圧縮されていない）レコードを保持していた期間が min.compaction.lag.ms 以上である場合、または (ii) ログがダーティな（圧縮されていない）レコードを保持していた期間が max.compaction.lag.ms 以内である場合。

型: double

デフォルト: 0.5

指定可能な値: [0,...,1]

サーバーのデフォルトプロパティ: log.cleaner.min.cleanable.ratio

重要度: 中
min.compaction.lag.ms

ログ内でメッセージが圧縮されずに保持される最小時間。圧縮中のログにのみ適用されます。

型: long

デフォルト: 0

指定可能な値: [0,...]

サーバーのデフォルトプロパティ: log.cleaner.min.compaction.lag.ms

重要度: 中
min.insync.replicas

プロデューサーが acks を "all"（または "-1"）に設定した場合、この構成パラメーターは、書き込みが正常に行われたとみなされるために、その書き込みを確認しなければならないレプリカの最小数を指定します。この最小数に達しない場合、プロデューサーは例外（NotEnoughReplicas または NotEnoughReplicasAfterAppend）を発します。
min.insync.replicas と acks を併用すると、より大きな永続性保証を適用できます。一般的には、トピックの作成時にレプリケーション係数を 3、min.insync.replicas を 2 に設定し、生成時の acks を "all" にします。これにより、大部分のレプリカが書き込みを受け取らなかった場合に、確実にプロデューサーから例外が生成されます。

型: int

デフォルト: 1

指定可能な値: [1,...]

サーバーのデフォルトプロパティ: min.insync.replicas

重要度: 中
preallocate

新しいログセグメントの作成時にファイルをディスクに事前割り当てする必要がある場合は、true です。

型: boolean

デフォルト: false

指定可能な値:

サーバーのデフォルトプロパティ: log.preallocate

重要度: 中
retention.bytes

この構成は、"delete" 保持ポリシーを使用している場合に、古いログセグメントを破棄してスペースを空ける前に、（ログセグメントから構成される）パーティションが到達できる最大サイズを制御します。デフォルトではサイズの制限はなく、時間の制限があるだけです。この制限はパーティションレベルで適用されるため、これにパーティションの数を掛けてトピック保持サイズを計算します（単位: バイト）。

型: long

デフォルト: -1

指定可能な値:

サーバーのデフォルトプロパティ: log.retention.bytes

重要度: 中
retention.ms

この構成は、"delete" 保持ポリシーを使用している場合に、古いログセグメントを破棄してスペースを空ける前に、ログを保持する最長時間を制御します。これは、コンシューマーがいかに早くデータを読み取る必要があるかに関する SLA に該当します。-1 に設定されている場合、時間制限は適用されません。

型: long

デフォルト: 604800000（7 日）

指定可能な値: [-1,...]

サーバーのデフォルトプロパティ: log.retention.ms

重要度: 中
segment.bytes

この構成はログのセグメントファイルサイズを制御します。保持およびクリーニングは常に 1 回 1 ファイルずつ行われるため、セグメントサイズを大きくするとファイルの数は少なくなりますが、保持を細かく制御できなくなります。

型: int

デフォルト: 1073741824（1 ギビバイト）

指定可能な値: [14,...]

サーバーのデフォルトプロパティ: log.segment.bytes

重要度: 中
segment.index.bytes

この構成は、オフセットをファイルの位置にマップするインデックスのサイズを制御します。このインデックスファイルは事前割り当てし、縮小するのはログのロール後のみです。通常は、この設定を変更する必要はありません。

型: int

デフォルト: 10485760（10 メビバイト）

指定可能な値: [0,...]

サーバーのデフォルトプロパティ: log.index.size.max.bytes

重要度: 中
segment.jitter.ms

セグメントローリングの thundering herd を回避するために、スケジュール設定されたセグメントロール時間から減算する最大ランダムジッター。

型: long

デフォルト: 0

指定可能な値: [0,...]

サーバーのデフォルトプロパティ: log.roll.jitter.ms

重要度: 中
segment.ms

この構成は、保持で古いデータを確実に削除または圧縮できるように、セグメントファイルが完全でなくても Kafka がログのロールを強制するまでの時間を制御します。

型: long

デフォルト: 604800000（7 日）

指定可能な値: [1,...]

サーバーのデフォルトプロパティ: log.roll.ms

重要度: 中
unclean.leader.election.enable

最終手段として、ISR セットにないレプリカをリーダーに選出できるようにするかどうかを示します。ただしこの場合、データ喪失の可能性があります。

型: boolean

デフォルト: false

指定可能な値:

サーバーのデフォルトプロパティ: unclean.leader.election.enable

重要度: 中
confluent.placement.constraints

この構成は、ISR への参加が常に許可されるブローカーのセット（レプリカ）を制御する JSON オブジェクトです。また、ISR への参加が常に許可されないブローカーのセット（オブザーバー）も制御できます。JSON の形式は次のとおりです。
{
"version": 1,
"replicas": [
{
"count": 2,
"constraints": {"rack": "east-1"}
},
{
"count": 1,
"constraints": {"rack": "east-2"}
}
],
"observers":[
{
"count": 1,
"constraints": {"rack": "west-1"}
}
]
}

型: string

デフォルト: ""

指定可能な値: kafka.common.TopicPlacement$TopicPlacementValidator@4c15e7fd

サーバーのデフォルトプロパティ: null

重要度: 低
message.downconversion.enable

この構成では、コンシューマーリクエストを満たすために、メッセージフォーマットのダウンコンバートを有効にするかどうかを制御します。false にすると、コンシューマーが以前のメッセージフォーマットを想定していても、ブローカーはダウンコンバートを実行しません。ブローカーは、このような以前のクライアントからのコンシューマーリクエストに対して UNSUPPORTED_VERSION エラーで応答します。この構成は、レプリケーションで準拠する必要のあるメッセージフォーマット規則には適用されません。

型: boolean

デフォルト: true

指定可能な値:

サーバーのデフォルトプロパティ: log.message.downconversion.enable

重要度: 低

注釈

このウェブサイトには、Apache License v2 の条件に基づいて Apache Software Foundation で開発されたコンテンツが含まれています。