KeyError：kafka.producer.record_accumulator.RecordBatch

Question

kafka-python apiを使ってトピックにたくさんのメッセージを送ります。成功したトピックに送信されますが、プログラムには、次のエラーメッセージを表示して終了する前にいないそれらのすべてが送信されたメッセージの一部：

KeyError: <kafka.producer.record_accumulator.RecordBatch object at 0x143d290> 
Batch is already closed -- ignoring batch.done() 
Error processing errback 
Traceback (most recent call last): 
    File "/usr/lib/python2.6/site-packages/kafka/future.py", line 79, in _call_backs 
    f(value) 
    File "/usr/lib/python2.6/site-packages/kafka/producer/sender.py", line 185, in _failed_produce 
    self._complete_batch(batch, error, -1, None) 
    File "/usr/lib/python2.6/site-packages/kafka/producer/sender.py", line 243, in _complete_batch 
    self._accumulator.deallocate(batch) 
    File "/usr/lib/python2.6/site-packages/kafka/producer/record_accumulator.py", line 507, in deallocate 
    self._incomplete.remove(batch) 
    File "/usr/lib/python2.6/site-packages/kafka/producer/record_accumulator.py", line 587, in remove 
    return self._incomplete.remove(batch) 

すべてのメッセージの異なる数を実行するには、実際に私のトピックで受信されています。この問題は、kafka producer.send呼び出しがプログラムが終了する前に送信を完了しないことが原因です。カフカドキュメンテーションのproducer.sendによる

は、おそらく根本的な原因である非同期メソッドである - すべての非同期スレッドを完了していないプロセスが殺される前に送信：

The send() method is asynchronous. When called it adds the record to a buffer of pending record sends and immediately returns. This allows the producer to batch together individual records for efficiency.

このためナイーブソリューションの数があります（たとえば、batch.sizeを低く設定するなど）、パフォーマンスのボトルネックが発生する可能性があります。

この問題を解決するにはどうすれば性能を損なうことなく、を解決しますか？

Answer 1

終了する前にproducer.flush()に電話するだけです。