メッセージ処理と冪等性の管理

Last updated on July 14, 2026

注釈：本資料はAI技術を用いて翻訳されています。

AccelByte Gaming Services (AGS) Extend Event Handler アプリのメッセージングシステムは、メッセージ配信の堅牢性、拡張性、耐障害性を確保するために Kafka エコシステムを採用しています。Extend は「at least once」配信セマンティクスに依存しており、これは Event Handler アプリが同じメッセージを少なくとも一度は受信することが保証されるという意味です。これにより、Event Handler やネットワークで例外やその他のエラーが発生しても、メッセージの再配信が試行されるため、メッセージが失われることはありません。

Extend は Kafka Connect を使用して Kafka トピックをサブスクライブし、データを取得し、メッセージを Event Handler の gRPC 呼び出しに変換します。ネットワークスループットを向上させるため、Kafka Connect は（設定可能な上限まで）バッチ単位でメッセージをポーリングおよび処理します。バッチが失敗した場合、そのバッチ全体が再試行されます。したがって、同じメッセージが複数回処理されることを許容できる、あるいはスムーズに処理できるようにするため、Event Handler は冪等性を持つ必要があります。

ユースケースによっては、同じアクションを複数回実行しても問題ない場合もあれば、メッセージの再送により 100 XP を複数回付与してしまうなど、問題になる場合もあります。冪等性を解決する最も簡単な方法は、イベント ID のような一意の識別子と分散データベース（例: Cloud Save）を使って、各メッセージの処理完了を一意に追跡することです。これにより、処理中に発生した失敗を取り消し、以降に見つかった重複メッセージを破棄できます。

例外とエラー処理

メッセージが Event Handler によって配信され、完全に処理されたことを保証する唯一の方法は、Event Handler からの確認応答（acknowledgement）を受け取ることです。Event Handler からの確認応答がない場合、Kafka Connect は、Event Handler または gRPC フレームワークが返す gRPC コードが「Retriable（再試行可能）」とみなされるステータスコードの集合に含まれるかどうかに応じて、バッチ全体の再配信を試みます。

Kafka Connect は、現在のバッチ内のすべてのメッセージが再試行不可（non-retryable）な gRPC ステータスコードで確認応答されるまで、新しいバッチのメッセージを送信しません。

成功、失敗、および再試行を通知するため（メッセージのバッチ全体が単位として再試行されます）、発生し得るすべての例外を捕捉して処理し、メッセージを処理できずかつ再試行を望む場合にのみ「Retriable」な gRPC ステータスコードをスローすることが重要です。

gRPC Event Handler から返される再試行可能な gRPC ステータスコードは以下のとおりです。

UNAVAILABLE: 例えば、Event Handler の gRPC サーバーが起動していない場合。
RESOURCE_EXHAUSTED: 例えば、サーバーが一時的にリソース不足になっている場合。
INTERNAL: 例えば、カスタムロジックの問題による内部サーバーエラーが発生した場合。
UNKNOWN: 例えば、Event Handler アプリが例外をスローした場合、または gRPC を終了する際にステータスコードを返す以外のアクションを実行した場合。

Kafka Connect の再試行回数を減らすには、今後の呼び出しで再試行によって解決し得る一時的な問題がある場合にのみ、再試行可能な gRPC ステータスコードを返してください。

C#
Go
Java
Python

bool allowFulfillment = true;
var fulfillmentCheck = _ABProvider.Sdk.Platform.Fulfillment.QueryFulfillmentHistoriesOp
    .SetUserId(request.UserId)
    .SetLimit(10)
    .SetOffset(0)
    .SetStatus(QueryFulfillmentHistoriesStatus.SUCCESS)
    .Execute(targetNamespace);
if (fulfillmentCheck != null && fulfillmentCheck.Data != null)
{
    foreach (var fulfillmentItem in fulfillmentCheck.Data)
    {
        if (fulfillmentItem.GrantedItemIds != null)
        {
            if (fulfillmentItem.GrantedItemIds.IndexOf(_ABProvider.ItemIdToGrant) > -1)
            {
                //item is already fulfilled for specified user
                allowFulfillment = false;
                break;
            }
        }
    }
}

if (allowFulfillment)
{
    var fulfillmentResponse = _ABProvider.Sdk.Platform.Fulfillment.FulfillItemOp
    .SetBody(new FulfillmentRequest()
    {
        ItemId = _ABProvider.ItemIdToGrant,
        Quantity = 1,
        Source = FulfillmentRequestSource.REWARD
    })
    .Execute(targetNamespace, request.UserId);
    if (fulfillmentResponse != null)
    {
        foreach (var entitlementItem in fulfillmentResponse.EntitlementSummaries!)
            _Logger.LogInformation($"EntitlementId: {entitlementItem.Id!}");
    }
}

func (o *LoginHandler) checkEntitlement(userID string, itemID string) (bool, error) {
	namespace := common.GetEnv("AB_NAMESPACE", "accelbyte")

	// Query fulfillment histories
	statusHistory := platformclientmodels.RequestHistoryStatusSUCCESS
	fulfillmentHistories, err := o.fulfillment.QueryFulfillmentHistoriesShort(&fulfillment.QueryFulfillmentHistoriesParams{
		Namespace: namespace,
		UserID:    &userID,
		Status:    &statusHistory,
	})
	if err != nil {
		return false, err
	}

	// Check if item has been granted
	if fulfillmentHistories != nil && fulfillmentHistories.Data != nil {
		for _, history := range fulfillmentHistories.Data {
			if history.GrantedItemIds != nil && len(history.GrantedItemIds) > 0 {
				for _, grantedItemID := range history.GrantedItemIds {
					if grantedItemID == itemID {
						return true, nil
					}
				}
			}
		}
	}

	return false, nil
}

func (o *LoginHandler) grantEntitlement(userID string, itemID string, count int32) error {
	namespace := common.GetEnv("AB_NAMESPACE", "accelbyte")

	// Fulfill the item
	fulfillmentResponse, err := o.fulfillment.FulfillItemShort(&fulfillment.FulfillItemParams{
		Namespace: namespace,
		UserID:    userID,
		Body: &platformclientmodels.FulfillmentRequest{
			ItemID:   itemID,
			Quantity: &count,
			Source:   platformclientmodels.EntitlementGrantSourceREWARD,
		},
	})
	if err != nil {
		return err
	}
	if fulfillmentResponse == nil || fulfillmentResponse.EntitlementSummaries == nil || len(fulfillmentResponse.EntitlementSummaries) <= 0 {
		return status.Errorf(codes.Internal, "could not grant item to user")
	}

	return nil
}

func (o *LoginHandler) OnMessage(ctx context.Context, msg *pb.UserLoggedIn) (*emptypb.Empty, error) {
	scope := common.GetScopeFromContext(ctx, "LoginHandler.OnMessage")
	defer scope.Finish()

	if itemIdToGrant == "" {
		return &emptypb.Empty{}, status.Errorf(codes.Internal, "Required envar ITEM_ID_TO_GRANT is not configured")
	}

	hasEntitlement, err := o.checkEntitlement(msg.UserId, itemIdToGrant)
	if err != nil {
		return &emptypb.Empty{}, status.Errorf(codes.Internal, "failed to check entitlement: %v", err)
	}
	if hasEntitlement {
		logrus.Infof("User already has the item, skipping...")
		
		return &emptypb.Empty{}, nil
	}

	err = o.grantEntitlement(msg.UserId, itemIdToGrant, 1)
	if err != nil {
		return &emptypb.Empty{}, status.Errorf(codes.Internal, "failed to grant entitlement: %v", err)
	}

	logrus.Infof("Entitlement granted successfully for user: %v", msg.UserId)
	
	return &emptypb.Empty{}, nil
}

private boolean checkEntitlement(String userId, String itemId) throws Exception {
    QueryFulfillmentHistories queryFulfillmentBody = QueryFulfillmentHistories.builder()
        .userId(userId)
        .limit(20)  // for our use case, checking the last 20 fulfillments is sufficient
        .offset(0)
        .status(QueryFulfillmentHistories.Status.SUCCESS.toString())
        .build();

    FulfillmentHistoryPagingSlicedResult fulfillmentCheck = fulfillment.queryFulfillmentHistories(queryFulfillmentBody);
    if (fulfillmentCheck != null && fulfillmentCheck.getData() != null) {
        for (FulfillmentHistoryInfo fulfillmentItem : fulfillmentCheck.getData()) {
            if (fulfillmentItem.getGrantedItemIds() != null) {
                if (fulfillmentItem.getGrantedItemIds().indexOf(itemId) > -1) {
                    // item is already fulfilled for specified user
                    return true;
                }
            }
        }
    }

    return false;
}

private void grantEntitlement(String userId, String itemId) throws Exception {
    FulfillmentRequest body = FulfillmentRequest.builder()
        .itemId(itemId)
        .quantity(1)
        .source(EntitlementGrant.Source.REWARD.name())
        .build();
    FulfillItem fulfillItemParam = FulfillItem.builder()
        .namespace(namespace)
        .userId(userId)
        .body(body)
        .build();
    FulfillmentResult fulfillmentResult = fulfillment.fulfillItem(fulfillItemParam);

    if (fulfillmentResult != null) {
        for (EntitlementSummary entitlementItem : fulfillmentResult.getEntitlementSummaries()) {
            log.info("entitlementId: {}", entitlementItem.getId());
        }
    }
}

@Override
public void onMessage(UserLoggedIn request, StreamObserver<Empty> responseObserver) {
    log.info("received a message: {}", request);
    String userId = request.getUserId();
    try {
        boolean hasEntitlement = checkEntitlement(userId, itemIdToGrant);
        if (hasEntitlement) {
            log.info("Item {} is already fulfilled, skipping", itemIdToGrant);
        } else {
            grantEntitlement(userId, itemIdToGrant);
        }

        responseObserver.onNext(Empty.getDefaultInstance());
        responseObserver.onCompleted();
    } catch (Exception e) {
        final String desc = String.format("could not grant item %s to user %s", itemIdToGrant, userId);
        log.error(desc, e);
        responseObserver.onError(Status.INTERNAL.withDescription(desc).asException());    // INTERNAL to retry
    }
}

def check_entitlement(self, user_id: str, item_id: str) -> bool:
    result, error = platform_service.query_fulfillment_histories(
        status="SUCCESS",
        user_id=user_id,
    )
    if not error and result.data:
        for fulfillment_history_info in result.data:
            if (
                hasattr(fulfillment_history_info, "granted_item_ids") and
                isinstance(fulfillment_history_info.granted_item_ids, list) and
                item_id in fulfillment_history_info.granted_item_ids
            ):
                return True
    return False

def grant_entitlement(
    self, user_id: str, item_id: str, count: int,
) -> Optional[Union[platform_models.ErrorEntity, HttpResponse]]:
    result, error = platform_service.fulfill_item(
        user_id=user_id,
        body=platform_models.FulfillmentRequest.create(
            quantity=count,
            item_id=item_id,
            source=platform_models.FulfillmentRequestSourceEnum.REWARD,
        )
    )
    if error:
        return error
    if len(result.entitlement_summaries) <= 0:
        raise Exception("could not grant item to user")
    return None

async def OnMessage(self, request: UserLoggedIn, context):
    self.log_payload(f"{self.OnMessage.__name__} request: %s", request)

    item_id = self.item_id_to_grant
    user_id = request.userId

    has_entitlement = self.check_entitlement(user_id=user_id, item_id=item_id)
    if has_entitlement:
        if self.logger:
            self.logger.info("user already has the item, skipping...")
        return Empty()

    error = self.grant_entitlement(request.userId, self.item_id_to_grant, 1)
    if error:
        error_msg = str(error.to_dict()) if isinstance(error, platform_models.ErrorEntity) else str(error)
        await context.abort(
            grpc.StatusCode.INTERNAL, f"could not grant item {item_id} to user {user_id}: {error_msg}",
        )  # INTERNAL to retry

    response = Empty()
    self.log_payload(f"{self.OnMessage.__name__} response: %s", response)
    return response

メッセージバッチの再試行期間は無期限であり、各バッチの再試行間隔は 30 秒に設定されています。すべての gRPC 呼び出しが再試行不可の戻りコードを受け取ると、バッチ全体のオフセットが Kafka にコミットされます。

参考情報

「at least once」セマンティクスを使用した処理保証と冪等性についての詳細は、Kafka Connect の公式ドキュメント（メッセージ配信保証について）を参照してください。

例外とエラー処理​

参考情報​

例外とエラー処理

参考情報