Flume-ngヌルイベントのカスタムシンク

Question

flume-ngのカスタムシンクを作成しようとしています。私は既存のシンクとドキュメンテーションを見て、それをコーディングしました。しかし、イベントを受け取るはずの 'process（）'メソッドは常にnullで終了します。 私はしているイベントイベント= channel.take（）;イベントはnullです。このメソッドは、イベントがまだチャンネルにあるときに繰り返し呼び出されます。

誰かが正しい方向に向かうことができますか？

Answer 1

これは、処理機能のスケルトンです...あなたがBACKOFFにステータスを変更し、をロールバックイベントを取得して失敗した場合。あなたでない場合をコミットし、ステータスをREADYに設定します。何があっても、あなたは常に取引を終了します。

Status status = null; 
    Channel channel = getChannel(); 
    Transaction transaction = channel.getTransaction(); 
    transaction.begin(); 
    try { 
     Event event = channel.take(); 

     if (event != null && validEvent(event.getBody()) >= 0) { 
      # make some printing 
     } 
     transaction.commit(); 
     status = Status.READY; 
    } catch (Throwable ex) { 
     transaction.rollback(); 
     status = Status.BACKOFF; 
     logger.error("Failed to deliver event. Exception follows.", ex); 
     throw new EventDeliveryException("Failed to deliver event: " + ex); 
    } finally { 
     transaction.close(); 
    } 
    return status; 

私はこれがうまくいくと確信しています:)。

Answer 2

これは仕様です。シンクランナーはシンクをnullイベントでポーリングするので、シンクが生きていて将来のイベントを受け入れる準備ができていることを確認できます。 nullイベントを受け取ったら、Status.BACKOFFを返すことを確認してから、シンクプロセッサーがもう一度お試しください。