public ControllerResult<Void> cleanupScaleInNodes() {
        List<ApiMessageAndVersion> records = new LinkedList<>();
        nodesMetadata.forEach((nodeId, nodeRuntimeMetadata) -> {
            if (clusterControlManager.isActive(nodeId) || cleaningUpNodes.contains(nodeId)) {
                return;
            }
            List<S3StreamSetObject> objects = new ArrayList<>(nodeRuntimeMetadata.streamSetObjects().values());
            boolean inMainStorageCircuitBreakerOpenStatus = objects.stream().anyMatch(sso -> {
                return Optional.ofNullable(s3ObjectControlManager.getObject(sso.objectId()))
                    .map(o -> ObjectAttributes.from(o.getAttributes()).bucket() == LocalFileObjectStorage.BUCKET_ID)
                    .orElse(false);
            });
            if (objects.isEmpty() || inMainStorageCircuitBreakerOpenStatus) {
                return;
            }
            CleanUpScaleInNodeContext ctx = new CleanUpScaleInNodeContext(nodeId, objects);
            cleaningUpNodes.add(nodeId);
            cleanupScaleInNode0(ctx);
            ctx.cf.whenComplete((nil, ex) -> {
                if (ex != null) {
                    log.error("cleanupScaleInNode failed", ex);
                }
                cleaningUpNodes.remove(nodeId);
            });

        });
        return ControllerResult.of(records, null);
    }
Check whether this node is valid to operate the stream related resources.
metadata/src/main/java/org/apache/kafka/controller/stream/StreamControlManager.java