Repository: apache/kafka
File: jmh-benchmarks/src/main/java/org/apache/kafka/jmh/assignor/StreamsStickyAssignorBenchmark.java (Java)

Code before:
        Map<String, StreamsGroupMember> members = createMembers();
        this.assignmentConfigs = Collections.singletonMap(
            "num.standby.replicas",
            Integer.toString(standbyReplicas)
        );
        this.groupSpec = StreamsAssignorBenchmarkUtils.createGroupSpec(members, assignmentConfigs);

        if (assignmentType == AssignmentType.INCREMENTAL) {
            simulateIncrementalRebalance();
        }
    }

    private Map<String, StreamsGroupMember> createMembers() {
        // In the rebalance case, we will add the last member as a trigger.
        // This is done to keep the total members count consistent with the input.
        int numberOfMembers = assignmentType.equals(AssignmentType.INCREMENTAL) ? memberCount - 1 : memberCount;

        return StreamsAssignorBenchmarkUtils.createStreamsMembers(
            numberOfMembers,
            membersPerProcess
        );
    }

    private void simulateIncrementalRebalance() {
        GroupAssignment initialAssignment = new StickyTaskAssignor().assign(groupSpec, topologyDescriber);
        Map<String, MemberAssignment> members = initialAssignment.members();

        Map<String, AssignmentMemberSpec> updatedMemberSpec = new HashMap<>();

        for (Map.Entry<String, AssignmentMemberSpec> member : groupSpec.members().entrySet()) {
            MemberAssignment memberAssignment = members.getOrDefault(
                member.getKey(),
                new MemberAssignment(Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap())
            );

            updatedMemberSpec.put(member.getKey(), new AssignmentMemberSpec(
                Optional.empty(),
                Optional.empty(),
                memberAssignment.activeTasks(),
                memberAssignment.standbyTasks(),
                memberAssignment.warmupTasks(),
                member.getValue().processId(),
                Map.of(),
                Map.of(),
                Map.of()
            ));
        }

        updatedMemberSpec.put("newMember", new AssignmentMemberSpec(
            Optional.empty(),
            Optional.empty(),

Review comment (suggestion):
Consider reusing the `taskAssignor` instance instead of creating a new `StickyTaskAssignor` here, to ensure consistent configuration and avoid unnecessary object creation.
```suggestion
        GroupAssignment initialAssignment = taskAssignor.assign(groupSpec, topologyDescriber);
```

Code after:

        Map<String, StreamsGroupMember> members = createMembers();
        this.assignmentConfigs = Collections.singletonMap(
            "num.standby.replicas",
            Integer.toString(standbyReplicas)
        );
        this.groupSpec = StreamsAssignorBenchmarkUtils.createGroupSpec(members, assignmentConfigs);

        if (assignmentType == AssignmentType.INCREMENTAL) {
            simulateIncrementalRebalance();
        }
    }

    private Map<String, StreamsGroupMember> createMembers() {
        // In the rebalance case, we will add the last member as a trigger.
        // This is done to keep the total members count consistent with the input.
        int numberOfMembers = assignmentType.equals(AssignmentType.INCREMENTAL) ? memberCount - 1 : memberCount;

        return StreamsAssignorBenchmarkUtils.createStreamsMembers(
            numberOfMembers,
            membersPerProcess
        );
    }

    private void simulateIncrementalRebalance() {
        GroupAssignment initialAssignment = new StickyTaskAssignor().assign(groupSpec, topologyDescriber);
        Map<String, MemberAssignment> members = initialAssignment.members();

        Map<String, AssignmentMemberSpec> updatedMemberSpec = new HashMap<>();

        for (Map.Entry<String, AssignmentMemberSpec> member : groupSpec.members().entrySet()) {
            MemberAssignment memberAssignment = members.getOrDefault(
                member.getKey(),
                new MemberAssignment(Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap())
            );

            updatedMemberSpec.put(member.getKey(), new AssignmentMemberSpec(
                Optional.empty(),
                Optional.empty(),
                memberAssignment.activeTasks(),
                memberAssignment.standbyTasks(),
                memberAssignment.warmupTasks(),
                member.getValue().processId(),
                Map.of(),
                Map.of(),
                Map.of()
            ));
        }

        updatedMemberSpec.put("newMember", new AssignmentMemberSpec(
            Optional.empty(),