void cleanupScaleInNode0(CleanUpScaleInNodeContext ctx) {
        if (ctx.index >= ctx.objects.size()) {
            ctx.cf.complete(null);
            return;
        }
        S3StreamSetObject object = ctx.objects.get(ctx.index);

        Optional<ObjectReader> objectReaderOpt = s3ObjectControlManager.objectReader(object.objectId());
        if (objectReaderOpt.isEmpty()) {
            ctx.cf.complete(null);
            return;
        }

        ObjectReader objectReader = objectReaderOpt.get();
        objectReader.basicObjectInfo().thenAccept(info -> {
            List<StreamOffsetRange> streamOffsetRanges = info.indexBlock().streamOffsetRanges();
            quorumController.appendWriteEvent("checkStreamSetObjectExpired", OptionalLong.empty(), () -> {
                return checkStreamSetObjectExpired(object, streamOffsetRanges);
            }).thenAccept(rst -> {
                // try clean up the node next object
                ctx.index = ctx.index + 1;
                quorumController.appendWriteEvent("cleanupScaleInNode0", OptionalLong.empty(), () -> {
                    cleanupScaleInNode0(ctx);
                    return ControllerResult.of(Collections.emptyList(), null);
                });
            }).exceptionally(ex -> {
                ctx.cf.completeExceptionally(ex);
                return null;
            }).whenComplete((nil, ex) -> {
                objectReader.release();
            });
        }).exceptionally(ex -> {
            ctx.cf.completeExceptionally(ex);
            return null;
        });
    }
Check whether this node is valid to operate the stream related resources.
metadata/src/main/java/org/apache/kafka/controller/stream/StreamControlManager.java