过度索取跟踪令牌

Question

我们注意到在将微服务扩展到 2 个副本时来自 TrackingEventProcessor class 的过多日志记录：

我们的 Axon 设置：

轴突版本 3.4.3
spring-引导版本 2.1.6.RELEASE
作为 TokenStore 的内部 couchbase 实现，即 CouchBaseTokenStore
PostgreSQL v11.7 作为事件存储
每个跟踪事件处理器的段数 = 1
跟踪事件处理器已配置forSingleThreadedProcessing

我们经常看到以下消息：

Segment is owned by another node. Releasing thread to process another segment...
No Worker Launcher active. Using current thread to assign segments.
Using current Thread for last segment worker: TrackingSegmentWorker{processor=core_policy-view, segment=Segment[0/0]}
Fetched token: GapAwareTrackingToken{index=2369007, gaps=[]} for segment: Segment[0/0]

这是一个名为 core_policy-view 的单个处理组的示例：

这是来自我们的暂存环境。我们在生产环境中没有看到相同的行为，即运行相同微服务的更多副本（5 个而不是 2 个），尽管是以前的版本。过多的日志记录值得关注，因为它会导致以下查询也针对事件存储过度执行：

SELECT min ( globalIndex ) FROM DomainEventEntry WHERE globalIndex > ?

下图显示了当微服务扩展到 2 个副本时它如何影响 APM：

通过查看 TrackingEventProcessor class 中的代码，我了解到 TrackingEventProcessor 负责将段分配给运行实例。如果该段已经被声明，那么一个特定的实例应该不会再麻烦了。然而，这似乎并没有发生。有时也会发生的是两个节点之间的 segment/token 乒乓球。

目前我什至可能没有正确阐明问题，而且我也不确定要问什么问题。如果有人能阐明我们是否做错了什么，我将不胜感激。我知道我们使用的 axon 版本是旧的，我们在升级路线图上有它，但现在我们需要将系统的当前更改投入生产，以便我们可以继续前进，然后开始升级.

编辑 1

以下是 CouchBaseTokenStore 中负责领取令牌的方法：

@Override
public void storeToken(TrackingToken token, String processorName, int segment) throws UnableToClaimTokenException {
    // Consider tracking CAS to avoid re-read and harness the CAS optimistic concurrency better.
    // Unfortunately, GapAwareTrackingToken is expected by RDBMS storage engines and can't be extended.
    JsonDocument doc = readOrCreateDocument(processorName, segment);
    if (GapAwareTrackingToken.class.isAssignableFrom(token.getClass())) {
        writeGapAwareToken((GapAwareTrackingToken) token, doc);
    } else {
        writeTrackingToken(token, doc);
    }
    this.axonStateBucket.upsert(doc);
}

@Override
public TrackingToken fetchToken(String processorName, int segment) throws UnableToClaimTokenException {
    JsonDocument doc = readOrCreateDocument(processorName, segment);
    String tokenClass = doc.content().getString(TOKEN_CLASS_FIELD);
    if (tokenClass == null) {
        return readGapAwareToken(doc);
    } else {
        return readTrackingToken(doc);
    }
}

private JsonDocument readOrCreateDocument(String processorName, int segment) throws UnableToClaimTokenException {
    String docId = getId(processorName, segment);
    JsonDocument doc = this.axonStateBucket.get(docId);
    if (doc == null) {
        try {
            doc = createDocument(processorName, segment);
        } catch (DocumentAlreadyExistsException e) {
            // Another instance beat us to it, read new token
            // which will most likely not be claimable.
            doc = this.axonStateBucket.get(docId);
        }
    }
    claimToken(doc);
    return doc;
}

private JsonDocument createDocument(String processorName, int segment) throws DocumentAlreadyExistsException {
    JsonObject content = JsonObject.create()
        .put(PROCESSOR_NAME_FIELD, processorName)
        .put(SEGMENT_FIELD, segment)
        .put(TYPE_FIELD, TOKEN_TYPE)
        .put(CLAIM_EXPIRY_FIELD, formatInstant(Instant.now().plus(claimDuration)))
        .put(OWNER_FIELD, nodeName);
    JsonDocument doc = JsonDocument.create(getId(processorName, segment), content);
    return this.axonStateBucket.insert(doc);
}

private void claimToken(JsonDocument document) throws UnableToClaimTokenException {
    String originalOwner = document.content().getString(OWNER_FIELD);
    Instant originalClaimExpiry = DateTimeUtils.parseInstant(document.content().getString(CLAIM_EXPIRY_FIELD));
    document.content()
        .put(CLAIM_EXPIRY_FIELD, formatInstant(Instant.now().plus(claimDuration)))
        .put(OWNER_FIELD, nodeName);
    if (nodeName.equals(originalOwner)) return;
    if ((originalClaimExpiry).isAfter(clock.instant())) {
        throw new UnableToClaimTokenException(String.format("Claim for owner %s is still valid.", originalOwner));
    }
}

Answer 1

在 Allard 的帮助下，我设法解决了这个问题（请参阅问题评论）。解决方法是在 fetch() 方法中声明令牌后也保留令牌。我们还开始使用 Couchbase SDK 提供的 replace() 方法而不是 upsert() 方法，以更好地利用 CAS（比较和交换）乐观并发：

@Override
public void storeToken(TrackingToken token, String processorName, int segment) throws UnableToClaimTokenException {
    JsonDocument doc = readOrCreateDocument(processorName, segment);
    if (GapAwareTrackingToken.class.isAssignableFrom(token.getClass())) {
        writeGapAwareToken((GapAwareTrackingToken) token, doc);
    } else {
        writeTrackingToken(token, doc);
    }
    axonStateBucket.replace(doc);
}

@Override
public TrackingToken fetchToken(String processorName, int segment) throws UnableToClaimTokenException {
    JsonDocument doc = readOrCreateDocument(processorName, segment);

    axonStateBucket.replace(doc); // readOrCreateDocument method claims, so we need to persist that

    String tokenClass = doc.content().getString(TOKEN_CLASS_FIELD);

    if (tokenClass == null) {
        return readGapAwareToken(doc);
    } else {
        return readTrackingToken(doc);
    }
}

其余代码与问题中的代码块相同。

过度索取跟踪令牌

Excessive claiming of tracking token

axon

axon-framework