private void handleAppendCallback0(WalWriteRequest request) {
        final long startTime = System.nanoTime();
        List<WalWriteRequest> waitingAckRequests;
        Lock lock = getStreamCallbackLock(request.record.getStreamId());
        lock.lock();
        try {
            waitingAckRequests = callbackSequencer.after(request);
            waitingAckRequests.forEach(r -> r.record.retain());
            for (WalWriteRequest waitingAckRequest : waitingAckRequests) {
                boolean full = deltaWALCache.put(waitingAckRequest.record);
                waitingAckRequest.confirmed = true;
                if (full) {
                    // cache block is full, trigger WAL upload.
                    uploadDeltaWAL();
                }
            }
        } finally {
            lock.unlock();
        }
        for (WalWriteRequest waitingAckRequest : waitingAckRequests) {
            waitingAckRequest.cf.complete(null);
        }
        StorageOperationStats.getInstance().appendCallbackStats.record(TimerUtil.timeElapsedSince(startTime, TimeUnit.NANOSECONDS));
    }
Force upload stream WAL cache to S3. Use group upload to avoid generate too many S3 objects when broker shutdown.
{@code streamId} can be {@link LogCache#MATCH_ALL_STREAMS} to force upload all streams.
s3stream/src/main/java/com/automq/stream/s3/S3Storage.java