Apache Kafka 3.1におけるメッセージフィルタリング：コンシューマー側での実装方法
Kafkaコンシューマーが大量の不要なメッセージを受信し、ネットワーク負荷や処理コストが増大する問題に直面したことはありませんか？プロデューサーコードを修正せずに、特定のデータを除外する方法を探していますか？本記事では、Apache Kafka 3.1のコンシューマーインターセプターを利用した、効率的なクライアントサイドでのメッセージフィルタリングを実装する方法を解説します。これにより、ネットワーク転送量と処理オーバーヘッドを削減し、システムのパフォーマンスを向上させることができます。
本記事を読み終えると、以下の知識を習得できるでしょう：
- Kafkaコンシューマーインターセプターの動作原理を理解する
- カスタムメッセージフィルタリングインターセプターの開発方法をマスターする
- インターセプターを設定してメッセージの選別を実現する方法を学ぶ
- メッセージフィルタリングのベストプラクティスと注意点を把握する
コンシューマーインターセプターの動作原理
Kafkaコンシューマーインターセプターは、プラグインインターフェースとして機能し、コンシューマーが受信するレコードをインターセプト（そして可能であれば変更）する機会を提供します。主に、サードパーティコンポーネントによるカスタム監視、ログ記録、またはメッセージフィルタリングなどの機能を実現するために使用されます。
コンシューマーインターセプターの核心となるインターフェースは、org.apache.kafka.clients.consumer.ConsumerInterceptorに定義されており、主に以下の3つのメソッドで構成されます：
onConsume(ConsumerRecords<K, V> records): アプリケーションにレコードが返される前に呼び出され、メッセージのフィルタリングや修正に使用できます。onCommit(Map<TopicPartition, OffsetAndMetadata> offsets): オフセットがコミットされる際に呼び出されます。close(): インターセプターがシャットダウンされる際に呼び出されます。
インターセプターは、KafkaConsumerのpoll()メソッド内部で呼び出され、org.apache.kafka.clients.consumer.internals.ConsumerInterceptorsクラスによって一元的に管理されます。
カスタムフィルタリングインターセプターの実装
ここでは、実際の例を通じてコンテンツベースのメッセージフィルタリングを実装する方法を紹介します。特定のキーワードを含むメッセージを除外するシナリオを想定します。
1. インターセプタークラスの作成
import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.Configurable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class ContentBasedFilterInterceptor implements ConsumerInterceptor<String, String> {
private List<String> blockedTerms;
@Override
public void configure(Map<String, ?> configs) {
// 設定からフィルタリング対象のキーワードを取得
String terms = (String) configs.get("message.filter.terms");
if (terms != null && !terms.isEmpty()) {
String[] termArray = terms.split(",");
blockedTerms = new ArrayList<>();
for (String term : termArray) {
blockedTerms.add(term.trim());
}
}
}
@Override
public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> inputRecords) {
if (blockedTerms == null || blockedTerms.isEmpty()) {
return inputRecords; // フィルタリングキーワードが設定されていない場合は、すべてのレコードをそのまま返す
}
Map<TopicPartition, List<org.apache.kafka.clients.consumer.ConsumerRecord<String, String>>> filteredRecordsMap = new HashMap<>();
// すべてのパーティションをループ処理
for (TopicPartition partition : inputRecords.partitions()) {
List<org.apache.kafka.clients.consumer.ConsumerRecord<String, String>> retainedRecords = new ArrayList<>();
// パーティション内の各レコードをチェック
for (org.apache.kafka.clients.consumer.ConsumerRecord<String, String> record : inputRecords.records(partition)) {
// メッセージ内容にフィルタリングキーワードが含まれているか確認
boolean shouldBlock = false;
if (record.value() != null) {
for (String term : blockedTerms) {
if (record.value().contains(term)) {
shouldBlock = true;
break;
}
}
}
// フィルタリングキーワードを含まない場合は、レコードを保持
if (!shouldBlock) {
retainedRecords.add(record);
}
}
if (!retainedRecords.isEmpty()) {
filteredRecordsMap.put(partition, retainedRecords);
}
}
return new ConsumerRecords<>(filteredRecordsMap);
}
@Override
public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
// オフセットコミット時の処理。必要に応じて実装。
}
@Override
public void close() {
// リソースのクリーンアップ。必要に応じて実装。
}
}
2. インターセプターのコンパイルとパッケージング
上記のコードをJARファイルにコンパイルし、Kafkaコンシューマーのクラスパスに含める必要があります。
インターセプターの設定と使用
1. コンシューマーの設定
コンシューマーの設定ファイルにインターセプタークラスと関連パラメータを追加します。
# consumer.properties
bootstrap.servers=localhost:9092
group.id=test-group
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
consumer.interceptor.classes=com.example.ContentBasedFilterInterceptor
message.filter.terms=error,debug,warning
設定パラメータの説明：
consumer.interceptor.classes: インターセプタークラスの完全修飾名を指定します。複数のインターセプターはカンマで区切ります。message.filter.terms: フィルタリング対象のキーワードをカンマ区切りで指定するカスタムパラメータです。
2. コンシューマーの起動
修正した設定ファイルを使用してコンシューマーを起動します。
bin/kafka-console-consumer.sh --consumer.config config/consumer.properties --topic sample-topic --from-beginning
動作フローと原理
インターセプターを追加した後のKafkaコンシューマーの処理フローは、ConsumerConfigクラス内のINTERCEPTOR_CLASSES_CONFIG設定によって決定されるインターセプターの呼び出し順序に従います。複数のインターセプターがチェーンとして機能し、一つのインターセプターの出力が次のインターセプターの入力となります。
高度な応用とベストプラクティス
1. 複雑なルールベースのフィルタリング
インターセプターを拡張して、より複雑なフィルタリングルールを実装できます。例えば、JSON形式のメッセージの特定フィールドを基にフィルタリングする場合：
// メッセージがJSON形式で、statusが"invalid"のものを除外する例
if (record.value() != null && record.value().contains(""status":"invalid"")) {
shouldBlock = true;
}
2. フィルタリングルールの動的更新
設定の再読み込みメカニズムを実装することで、コンシューマーを再起動せずにフィルタリングルールを更新できます。
@Override
public void configure(Map<String, ?> configs) {
// 外部の設定サービスから最新のフィルタリングルールを定期的に取得するスケジュールタスクを初期化
ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
scheduler.scheduleAtFixedRate(() -> {
blockedTerms = fetchLatestTermsFromConfigServer();
}, 0, 5, TimeUnit.MINUTES);
}
3. パフォーマンスに関する注意点
- フィルタリングロジックは可能な限りシンプルに保ち、インターセプター内での重い処理は避けてください。
- 大量のキーワードフィルタリングには、ブルームフィルタ（Bloom Filter）の使用を検討してください。
- インターセプターはコンシューマーのpollスレッド内で実行されるため、過度な処理はコンシューマーのパフォーマンスに影響を与えます。
よくある問題と解決策
問題1：インターセプターが動作しない
以下の点を確認してください：
- インターセプタークラスがクラスパスに含まれているか。
- 設定パラメータ
consumer.interceptor.classesが正しく設定されているか。 - インターセプターが
ConsumerInterceptorインターフェースを正しく実装しているか。 - コンシューマーログを確認し、インターセプターがスローした例外が記録されているか（例外は伝播されません）。
問題2：メッセージフィルタリングによるオフセットコミットの問題
解決策：
- インターセプターはアプリケーションに返されるメッセージのみをフィルタリングし、Kafkaのオフセットコミットには影響しません。
- 特定のメッセージを完全にスキップし、オフセットをコミットしたくない場合は、
seek()メソッドを使用してオフセットを手動で制御することを検討してください。