@Override
    ShareFetchResponse throttleResponse(int throttleTimeMs) {
        if (!isSubsequent) {
            return new ShareFetchResponse(ShareFetchResponse.toMessage(Errors.NONE, throttleTimeMs,
                    Collections.emptyIterator(), Collections.emptyList()));
        }
        int expectedEpoch = ShareFetchMetadata.nextEpoch(reqMetadata.epoch());
        int sessionEpoch;
        synchronized (session) {
            sessionEpoch = session.epoch;
        }
        if (sessionEpoch != expectedEpoch) {
            log.debug("Subsequent share session {} expected epoch {}, but got {}. " +
                    "Possible duplicate request.", session.key(), expectedEpoch, sessionEpoch);
            return new ShareFetchResponse(ShareFetchResponse.toMessage(Errors.INVALID_SHARE_SESSION_EPOCH,
                    throttleTimeMs, Collections.emptyIterator(), Collections.emptyList()));
        }
        return new ShareFetchResponse(ShareFetchResponse.toMessage(Errors.NONE, throttleTimeMs,
                Collections.emptyIterator(), Collections.emptyList()));
    }
The share fetch context for a subsequent request that utilizes an existing share session.

@param reqMetadata  The request metadata.
@param session      The subsequent fetch request session.
core/src/main/java/kafka/server/share/ShareSessionContext.java