メインコンテンツまでスキップ
に最終更新

承諾コールバック

確認コールバックを使用すると、クライアントはプロデューサーのループをブロックすることなく、レコードの確認応答やエラーに非同期で対応できます。レコードが永続化されるか、または失敗すると、Zerobus Ingest はバックグラウンドでコールバックを呼び出します。これにより、プロデューサーを遅延させることなく進捗状況の追跡やメトリクスの更新が可能になり、障害が発生した際に即座に把握できます。

これは オフセットの待機やフラッシュ とは異なります。これらは、コードがインラインで耐久性を待機するブロッキング呼び出しです。コールバックはブロッキング呼び出しではありません。これは、確認応答が到着したときに SDK が代わりに呼び出すハンドラです。

JSON および Protocol Buffers (protobuf) SDK ストリームでは、確認コールバックがサポートされています。Arrow Flight ストリームはコールバックをサポートしていません。Arrow ストリームの耐久性を確認するには、wait_for_offset() または flush() を使用してください。Zerobus Ingest での Arrow Flight の使用を参照してください。

以下のメソッド名と型名は、Python SDKのものです。その他のZerobus SDKでは、サポートされている場合、各言語の同等の構成を使用して、確認コールバックを公開します。

コールバックの仕組み​

AckCallback をサブクラス化し、2 つのメソッドを実装することでコールバックを定義します。

  • on_ack(offset: int)：送信（レコードまたはバッチ）がサーバーによって永続的であると正常に承認されたときに呼び出されます。offsetは、承認された送信を識別します。
  • on_error(offset: int, error_message: str)：送信でエラーが発生したときに呼び出されます。on_error は任意です。障害を処理または Logs に記録するために実装します。

コールバックは、送信された各レコードまたはバッチの論理オフセットが確認または失敗したときに 1 回呼び出されるため、ストリーム全体にわたる取り込みの進行状況を示す実行中のシグナルとなります。

コールバックメソッドは SDK のバックグラウンドスレッドで実行されるため、それらを呼び出してもプロデューサーはブロックされません。それらを高速かつ非ブロックの状態に保ってください。障害発生時の対応はクライアントの責任となります：log、アラート、再試行、または停止。一部のエラーは致命的であり、on_errorがストリームの永続的な失敗を報告した場合は、新しいストリームでリカバリする必要があります。リカバリおよび再試行パターンを参照してください。

コールバックの構成​

ストリームを作成する際に、AckCallback サブクラスのインスタンスを StreamConfigurationOptions の ack_callback オプションとして渡すことで、ストリームにコールバックをアタッチします。その後、コールバックはそのストリームに取り込まれたすべてのレコードに適用されます。

Python
from zerobus.sdk.sync import ZerobusSdk
from zerobus.sdk.shared import AckCallback, StreamConfigurationOptions, TableProperties

class MyAckCallback(AckCallback):
    def on_ack(self, offset: int) -> None:
        print(f"Record acknowledged at offset: {offset}")

    def on_error(self, offset: int, error_message: str) -> None:
        print(f"Error at offset {offset}: {error_message}")

sdk = ZerobusSdk(SERVER_ENDPOINT, DATABRICKS_WORKSPACE_URL)

table_properties = TableProperties("main.default.air_quality")
options = StreamConfigurationOptions(
    ack_callback=MyAckCallback(),
)
stream = sdk.create_stream(CLIENT_ID, CLIENT_SECRET, table_properties, options)

try:
    for row in records:
        stream.ingest_record_offset(row)
finally:
    stream.close()

コールバックが登録されているため、インラインで待機する必要はありません。各レコードの永続性が確認されると on_ack が発生し、レコードが失敗すると on_error が発生します。

コールバック対ブロッキング​

コールバックとブロッキング呼び出しはそれぞれ異なる問題を解決するものであり、これらを併用することも可能です：

  • 高throughputを維持しながら、耐久性の確認やエラーが発生した際に非同期で対応するには、確認コールバックを使用します。進捗の追跡、メトリクス、エラーログ記録に適しています。
  • コードが続行する前に特定のレコード、またはすべての保留中のレコードが永続化されるまでブロックする必要がある場合は、wait_for_offset() または flush() を使用してください。

関連​

このページの見出し