Hortonworksが提供するカスタム Flume ソースとFlume Kafka シンクを介してメッセージを生成し、 Kafka ブローカーに書き込むプロセスを作成しました。
このプロセス中に、KAFKA ブローカーが既に実行されている場合に Flume エージェントを起動すると、すべてのメッセージが Kafka ブローカーに適切に配信されますが、Flume エージェントが既に実行されているときに Kafka ブローカーを起動すると、KAFKA ブローカーはすべてのメッセージを受け取ります。Kafka コンソール コンシューマーを
実行して受信したメッセージの数を確認すると、最初からいくつかのレコードがドロップされ、最後からいくつかのレコードがドロップされていることに気付きました。
Flume.confで複数の組み合わせを試しましたが、それでも期待どおりに動作しています。
以下は、 Flume.confに提供した構成パラメーターです。
agent.channels = firehose-channel
agent.sources = stress-source
agent.sinks = kafkasink
#################################
# Benchmark Souce Configuration #
#################################
agent.sources.stress-source.type=com.kohls.flume.source.stress.BenchMarkTestScenriao
agent.sources.stress-source.size=5000
agent.sources.stress-source.maxTotalEvents=30000
agent.sources.stress-source.batchSize=200
agent.sources.stress-source.throughputThreshold=4000
agent.sources.stress-source.throughputControlSeconds=1
agent.sources.stress-source.channels=firehose-channel
#################################
# Firehose Channel Configuration #
#################################
agent.channels.firehose-channel.type = file
agent.channels.firehose-channel.checkpointDir = /data/flume/checkpoint
agent.channels.firehose-channel.dataDirs = /data/flume/data
agent.channels.firehose-channel.capacity = 10000
agent.channels.firehose-channel.transactionCapacity = 10000
agent.channels.firehose-channel.useDualCheckpoints=1
agent.channels.firehose-channel.backupCheckpointDir=/data/flume/backup
############################################
# Firehose Sink Configuration - Kafka Sink #
############################################
agent.sinks.kafkasink.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.kafkasink.topic = backoff_test_17
agent.sinks.kafkasink.channel=firehose-channel
agent.sinks.kafkasink.brokerList = sandbox.hortonworks.com:6667
agent.sinks.kafkasink.batchsize = 200
agent.sinks.kafkasink.requiredAcks = 1
agent.sinks.kafkasink.kafka.producer.type = async
agent.sinks.kafkasink.kafka.batch.num.messages = 200
また、flume ログの分析を試みたところ、flume メトリクスがPUTおよびTAKEカウントを適切に示していることがわかりました。
この問題を解決するためのポインタがあれば教えてください。
あらかじめご了承ください。