Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java (Java)

Code before:
                                final Map<String, Subtopology> subtopologies,
                                final Map<String, String> clientTags) {
        this.processId = Objects.requireNonNull(processId, "Process ID cannot be null");
        this.endpoint = Objects.requireNonNull(endpoint, "Endpoint cannot be null");
        this.subtopologies = Map.copyOf(Objects.requireNonNull(subtopologies, "Subtopologies cannot be null"));
        this.clientTags = Map.copyOf(Objects.requireNonNull(clientTags, "Client tags cannot be null"));
    }

    public UUID processId() {
        return processId;
    }

    public Optional<HostInfo> endpoint() {
        return endpoint;
    }

    public Map<String, String> clientTags() {
        return clientTags;
    }

    public Map<String, Subtopology> subtopologies() {
        return subtopologies;
    }

    public int endpointInformationEpoch() {
        return endpointInformationEpoch;
    }
    public int topologyEpoch() {
        return 0;
    }

    public void setReconciledAssignment(final Assignment assignment) {
        reconciledAssignment.set(assignment);
    }

    public Assignment reconciledAssignment() {
        return reconciledAssignment.get();
    }

    public void setPartitionsByHost(final Map<HostInfo, EndpointPartitions> partitionsByHost) {
        this.partitionsByHost.set(partitionsByHost);
    }

    public void setEndpointInformationEpoch(final int endpointInformationEpoch) {
        this.endpointInformationEpoch = endpointInformationEpoch;
    }

    public Map<HostInfo, EndpointPartitions> partitionsByHost() {
        return partitionsByHost.get();
    }


Review comment (suggestion):
`StreamsRebalanceData` is used as an interface between the application thread and the streams thread. The data here should be immutable or otherwise threadsafe. But I'm not sure we need even access this epoch from the application thread? I'd consider moving the client-side epoch to `StreamsGroupHeartbeatManager.HeartbeatState`

Code after:
        this.processId = Objects.requireNonNull(processId, "Process ID cannot be null");
        this.endpoint = Objects.requireNonNull(endpoint, "Endpoint cannot be null");
        this.subtopologies = Map.copyOf(Objects.requireNonNull(subtopologies, "Subtopologies cannot be null"));
        this.clientTags = Map.copyOf(Objects.requireNonNull(clientTags, "Client tags cannot be null"));
    }

    public UUID processId() {
        return processId;
    }

    public Optional<HostInfo> endpoint() {
        return endpoint;
    }

    public Map<String, String> clientTags() {
        return clientTags;
    }

    public Map<String, Subtopology> subtopologies() {
        return subtopologies;
    }

    public int topologyEpoch() {
        return 0;
    }

    public void setReconciledAssignment(final Assignment assignment) {
        reconciledAssignment.set(assignment);
    }

    public Assignment reconciledAssignment() {
        return reconciledAssignment.get();
    }

    public void setPartitionsByHost(final Map<HostInfo, EndpointPartitions> partitionsByHost) {
        this.partitionsByHost.set(partitionsByHost);
    }

    public Map<HostInfo, EndpointPartitions> partitionsByHost() {
        return partitionsByHost.get();
    }

    /** For the current stream thread to request a shutdown of all Streams clients belonging to the same application. */
    public void requestShutdown() {
        shutdownRequested.set(true);
    }

    /** True if the current stream thread requested a shutdown of all Streams clients belonging to the same application. */
    public boolean shutdownRequested() {
        return shutdownRequested.get();
    }