プラットフォームイベント開発者ガイド
j検出されなかった例外発生後のプラットフォームイベントトリガーの再開
処理するイベントメッセージ数が少ないほど、トリガーは Apex ガバナ制限に達する可能性が低くなります。プラットフォームイベントトリガーの最大バッチサイズは 2,000 で、Apex オブジェクトトリガーの最大バッチサイズは 200 です。そのため、プラットフォームイベントトリガーは制限に達する可能性がより高く、この機能によるメリットがあります。
トリガー再開のチェックポイントを設定するには、最後に正常に処理されたイベントメッセージの再実行 ID をこのメソッドのコールを使用して設定します。
1EventBus.TriggerContext.currentContext().setResumeCheckpoint(replayId);トリガーの実行フローが意図的に、または未処理の例外 (制限の例外など) によって停止すると、新しいバッチ (Trigger.New の sObject リスト) を使用して、トリガーがもう一度起動します。新しいバッチは、設定した再実行 ID の後のイベントメッセージから開始します。イベント��、変化しない ReplayID 項目値に基づいて、最初の順番で再送信されます。トリガーにより、再送信されるイベントと以降のバッチが順次処理されます。setResumeCheckpoint(replayId) メソッドによってトリガー実行は停止しませんが、明示的に実行を終了することができます。たとえば、バッチサイズを制御するには、一部のイベントメッセージが処理されたら実行フローを終了するようにします。
指定された再生 ID が有効でない場合、このメソッドでは EventBus.InvalidReplayIdException が発生します。無効な再生 ID は、Trigger.new リストのイベントの現在のトリガーバッチにない再生 ID です。
例
このトリガー例では、各反復の最後に処理されたイベントメッセージの再実行 ID を設定します。制限の例外が発生すると、トリガーがもう一度起動し、設定された再実行 ID の後のイベントメッセージから処理が再開されます。
1trigger ResumeEventProcessingTrigger on Low_Ink__e (after insert) {
2 for (Low_Ink__e event : Trigger.New) {
3 // Process the event message.
4 // ...
5
6 // Set the Replay ID of the last successfully processed event message.
7 // If a limit is hit, the trigger refires and processing starts with the
8 // event after the last one processed (the set Replay ID).
9 EventBus.TriggerContext.currentContext().setResumeCheckpoint(event.replayId);
10 }
11}例
この例では、プラットフォームイベントトリガーのバッチサイズを制御し、Apex オブジェクトトリガーのバッチサイズ 200 に一致させます。このトリガーは、すでに処理済みのイベントメッセージの数を数えます。setResumeCheckpoint(replayId) は、正常に処理された各イベントメッセージの後でループの各反復でコールされます。イベント数が 200 件を超えるとループが終了し、トリガーの実行が停止されます。未処理のイベントメッセージがある場合、トリガーは再度起動します。新しいトリガー呼び出しに送信されたイベントメッセージのリストは、設定された再実行 ID のイベントメッセージの後のイベントメッセージで開始されます。
1trigger ControlBatchSizeTrigger on Low_Ink__e (after insert) {
2 Integer counter = 0;
3 for (Low_Ink__e event : Trigger.New) {
4 // Increase batch counter.
5 counter++;
6 // Only process the first 200 event messages
7 if (counter > 200) {
8 // Resume after the last successfully processed event message
9 // after the trigger stops running.
10 // Exit for loop.
11 break;
12 }
13
14 // Process event message.
15 // ....
16
17 // Set Replay ID after which to resume event processing
18 // in new trigger execution.
19 EventBus.TriggerContext.currentContext().setResumeCheckpoint(
20 event.ReplayId);
21 }
22}TestBatchSizeTriggerResumption テストクラスには ControlBatchSizeTrigger のテストが含まれます。クラスのテストメソッドは 201 件のイベントメッセージを公開します。次に、deliver() メソッドを 2 回コールしてトリガーを 2 回起動します。最初の呼び出しは、200 件のイベントメッセージを処理します。2 番目の呼び出しは、最後のイベントメッセージを処理します。このテス���では、最後に処理されたイベントメッセージの再実行 ID を保持する EventBusSubscriber.Position プロパティを調査してトリガーが呼び出されたことを検証します。
1@isTest
2public class TestBatchSizeTriggerResumption {
3
4 @isTest static void testResumingBatchSizeTrigger() {
5
6 Test.startTest();
7
8 // Publish 201 test events
9 List<Low_Ink__e> eventList = new List<Low_Ink__e>();
10 for(Integer i=0;i<201;i++) {
11 Low_Ink__e oneEvent = new Low_Ink__e(Serial_Number__c='X-' + i);
12 eventList.add(oneEvent);
13 }
14 Database.SaveResult[] srs = EventBus.publish(eventList);
15 for(Database.SaveResult sr : srs) {
16 System.assertEquals(true, sr.isSuccess());
17 }
18
19
20 // Deliver the first 200 test event messages.
21 // This will fire the associated event trigger.
22 Test.getEventBus().deliver();
23
24 // Get old position of this subscriber
25 EventBusSubscriber subOld =
26 [SELECT Name, Position, Topic
27 FROM EventBusSubscriber
28 WHERE Topic='Low_Ink__e' AND Name='ControlBatchSizeTrigger'];
29 System.debug(subOld);
30
31 // Refire the trigger for the last event (201st).
32 Test.getEventBus().deliver();
33
34 // VERIFICATION
35 // Get new position of this subscriber
36 EventBusSubscriber subNew =
37 [SELECT Name, Position, Topic
38 FROM EventBusSubscriber
39 WHERE Topic='Low_Ink__e' AND Name='ControlBatchSizeTrigger'];
40 System.debug(subNew);
41
42 System.assertEquals(subOld.Position + 1, subNew.Position);
43
44 Test.stopTest();
45
46 }
47}