private void tryReassignPartitionBack(StreamRuntimeMetadata stream) {
        ControllerRequestContext context = new ControllerRequestContext(null, null, OptionalLong.empty());
        AlterPartitionReassignmentsRequestData request = new AlterPartitionReassignmentsRequestData();
        String rawTopicId = stream.tags().get(StreamTags.Topic.KEY);
        String rawPartitionIndex = stream.tags().get(StreamTags.Partition.KEY);
        if (Strings.isNullOrEmpty(rawTopicId) || Strings.isNullOrEmpty(rawPartitionIndex)) {
            return;
        }
        Uuid topicId = Uuid.fromString(rawTopicId);
        int partitionIndex = StreamTags.Partition.decode(rawPartitionIndex);
        int nodeId = stream.currentRangeOwner();
        quorumController.findTopicNames(context, List.of(topicId)).thenAccept(uuid2name -> {
            String topicName = Optional.ofNullable(uuid2name.get(topicId)).filter(r -> !r.isError()).map(r -> r.result()).orElse(null);
            if (topicName == null) {
                return;
            }
            request.setTopics(List.of(new AlterPartitionReassignmentsRequestData.ReassignableTopic()
                .setName(topicName)
                .setPartitions(List.of(
                    new AlterPartitionReassignmentsRequestData.ReassignablePartition()
                        .setPartitionIndex(partitionIndex)
                        .setReplicas(List.of(nodeId))
                ))));
            quorumController.alterPartitionReassignments(context, request)
                .thenAccept(rst -> {
                    LOGGER.info("[REASSIGN_PARTITION_BACK_TO_LOCKED_NODE],req={},resp={}", request, rst);
                });
        });
    }
Check whether this node is valid to operate the stream related resources.
metadata/src/main/java/org/apache/kafka/controller/stream/StreamControlManager.java