// From GroupCoordinatorService.java:394
public int partitionFor(String groupId) {
    return Utils.abs(groupId.hashCode()) % numPartitions;
}

public CoordinatorResult<Void, CoordinatorRecord> classicGroupJoin(
    AuthorizableRequestContext context,
    JoinGroupRequestData request,
    CompletableFuture<JoinGroupResponseData> responseFuture
) {
    Group group = groups.get(request.groupId(), Long.MAX_VALUE);
    if (group != null) {
        if (group.type() == CONSUMER && !group.isEmpty()) {
            return classicGroupJoinToConsumerGroup((ConsumerGroup) group, context, request, responseFuture);
        } else if (group.type() == CONSUMER || group.type() == CLASSIC || group.type() == STREAMS && group.isEmpty()) {
            return classicGroupJoinToClassicGroup(context, request, responseFuture);
        } else {
            responseFuture.complete(new JoinGroupResponseData()
                .setMemberId(UNKNOWN_MEMBER_ID)
                .setErrorCode(Errors.INCONSISTENT_GROUP_PROTOCOL.code())
            );
            return EMPTY_RESULT;
        }
    } else {
        return classicGroupJoinToClassicGroup(context, request, responseFuture);
    }
}

public CoordinatorResult<Void, CoordinatorRecord> classicGroupSync(
    AuthorizableRequestContext context,
    SyncGroupRequestData request,
    CompletableFuture<SyncGroupResponseData> responseFuture
) {
    Group group;
    try {
        group = group(request.groupId());
    } catch (GroupIdNotFoundException e) {
        responseFuture.complete(new SyncGroupResponseData()
            .setErrorCode(Errors.UNKNOWN_MEMBER_ID.code()));
        return EMPTY_RESULT;
    }

    if (group.type() == CLASSIC) {
        return classicGroupSyncToClassicGroup((ClassicGroup) group, context, request, responseFuture);
    } else if (group.type() == CONSUMER) {
        return classicGroupSyncToConsumerGroup((ConsumerGroup) group, request, responseFuture);
    } else {
        responseFuture.complete(new SyncGroupResponseData()
            .setErrorCode(Errors.UNKNOWN_MEMBER_ID.code()));
        return EMPTY_RESULT;
    }
}

CoordinatorResult<Void, CoordinatorRecord> prepareRebalance(
    ClassicGroup group,
    String reason
) {
    // If any members are awaiting sync, cancel their request and have them rejoin.
    if (group.isInState(COMPLETING_REBALANCE)) {
        resetAndPropagateAssignmentWithError(group, Errors.REBALANCE_IN_PROGRESS);
    }

    // If a sync expiration is pending, cancel it.
    removeSyncExpiration(group);

    group.transitionTo(PREPARING_REBALANCE);

    log.info("Preparing to rebalance group {} in state {} with old generation {} (reason: {}).",
        group.groupId(),
        group.currentState(),
        group.generationId(),
        reason);

    return EMPTY_RESULT;
}

// From ClassicGroup.java:1289
public void initNextGeneration() {
    generationId++;
    if (!members.isEmpty()) {
        setProtocolName(Optional.of(selectProtocol()));
        subscribedTopics = computeSubscribedTopics();
        transitionTo(COMPLETING_REBALANCE);
    } else {
        setProtocolName(Optional.empty());
        subscribedTopics = computeSubscribedTopics();
        transitionTo(EMPTY);
    }
    clearPendingSyncMembers();
}

// Simplified algorithm (actual implementation includes rack-aware assignment):
// 1. For each topic independently:
//    - Sort consumers lexicographically
//    - Sort partitions numerically
//    - Divide partitions evenly: numPartitions / numConsumers
//    - First (numPartitions % numConsumers) consumers get one extra partition

public Map<String, List<TopicPartition>> assign(
    Map<String, Integer> partitionsPerTopic,
    Map<String, Subscription> subscriptions) {
    return assignPartitions(partitionInfosWithoutRacks(partitionsPerTopic), subscriptions);
}

// Simplified algorithm:
// 1. Sort all partitions across all topics alphabetically
// 2. Sort all consumers alphabetically
// 3. Use circular iterator to assign partitions round-robin to consumers
//    (skipping consumers not subscribed to that partition's topic)

public Map<String, List<TopicPartition>> assign(
    Map<String, Integer> partitionsPerTopic,
    Map<String, Subscription> subscriptions) {

    CircularIterator<MemberInfo> assigner = new CircularIterator<>(Utils.sorted(memberInfoList));

    for (TopicPartition partition : allPartitionsSorted(partitionsPerTopic, subscriptions)) {
        // Skip to next consumer subscribed to this topic
        while (!subscriptions.get(assigner.peek().memberId).topics().contains(partition.topic()))
            assigner.next();
        assignment.get(assigner.next().memberId).add(partition);
    }
    return assignment;
}

// Simplified algorithm:
// 1. Extract previous partition assignments from member metadata (owned partitions)
// 2. If all consumers subscribe to same topics:
//    - Use ConstrainedAssignmentBuilder (optimized algorithm)
//    - Preserves existing assignments where possible
// 3. Otherwise:
//    - Use GeneralAssignmentBuilder
//    - Compute sticky assignment from scratch
// 4. Balance unassigned partitions across consumers while minimizing movement

public Map<String, List<TopicPartition>> assign(
    Map<String, Integer> partitionsPerTopic,
    Map<String, Subscription> subscriptions) {

    AbstractAssignmentBuilder assignmentBuilder;
    if (allSubscriptionsEqual(partitionsPerTopic.keySet(), subscriptions, /*...*/)) {
        assignmentBuilder = new ConstrainedAssignmentBuilder(/* preserve assignments */);
    } else {
        assignmentBuilder = new GeneralAssignmentBuilder(/* recompute */);
    }
    return assignmentBuilder.build();
}

switch (protocol) {
    case EAGER:
        // revoke all partitions
        revokedPartitions.addAll(subscriptions.assignedPartitions());
        exception = rebalanceListenerInvoker.invokePartitionsRevoked(revokedPartitions);
        subscriptions.assignFromSubscribed(Collections.emptySet());
        break;

    case COOPERATIVE:
        // only revoke those partitions that are not in the subscription anymore.
        Set<TopicPartition> ownedPartitions = new HashSet<>(subscriptions.assignedPartitions());
        revokedPartitions.addAll(ownedPartitions.stream()
            .filter(tp -> !subscriptions.subscription().contains(tp.topic()))
            .collect(Collectors.toSet()));

        if (!revokedPartitions.isEmpty()) {
            exception = rebalanceListenerInvoker.invokePartitionsRevoked(revokedPartitions);
            ownedPartitions.removeAll(revokedPartitions);
            subscriptions.assignFromSubscribed(ownedPartitions);
        }
        break;
}

public CoordinatorResult<OffsetCommitResponseData, CoordinatorRecord> commitOffset(
    AuthorizableRequestContext context,
    OffsetCommitRequestData request
) throws ApiException {
    Group group = validateOffsetCommit(context, request);

    // In the old consumer group protocol, the offset commits maintain the session if
    // the group is in Stable or PreparingRebalance state.
    if (group.type() == Group.GroupType.CLASSIC) {
        ClassicGroup classicGroup = (ClassicGroup) group;
        if (classicGroup.isInState(STABLE) || classicGroup.isInState(PREPARING_REBALANCE)) {
            groupMetadataManager.rescheduleClassicGroupMemberHeartbeat(
                classicGroup,
                classicGroup.member(request.memberId())
            );
        }
    }

    final OffsetCommitResponseData response = new OffsetCommitResponseData();
    final List<CoordinatorRecord> records = new ArrayList<>();
    final long currentTimeMs = time.milliseconds();
    final OptionalLong expireTimestampMs = expireTimestampMs(request.retentionTimeMs(), currentTimeMs);

    request.topics().forEach(topic -> {
        topic.partitions().forEach(partition -> {
            // Build offset commit record
            final OffsetAndMetadata offsetAndMetadata = OffsetAndMetadata.fromRequest(
                topic.topicId(),
                partition,
                currentTimeMs,
                expireTimestampMs
            );

            // Add record to be written to __consumer_offsets
            records.add(CoordinatorRecordHelpers.newOffsetCommitRecord(
                request.groupId(),
                topic.name(),
                partition.partitionIndex(),
                offsetAndMetadata,
                metadataImage.features().metadataVersion()
            ));
        });
    });

    return new CoordinatorResult<>(records, response);
}

// Consumer config
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "5000");  // Commit every 5 seconds

// Consumer loop (simplified)
while (true) {
  ConsumerRecords<K, V> records = consumer.poll(Duration.ofMillis(100));

  for (ConsumerRecord<K, V> record : records) {
    processRecord(record);
  }

  // Auto-commit happens in background
  // - Triggered every auto.commit.interval.ms
  // - Commits offsets from last poll()
}

props.put("enable.auto.commit", "false");

while (true) {
  ConsumerRecords<K, V> records = consumer.poll(Duration.ofMillis(100));

  for (ConsumerRecord<K, V> record : records) {
    processRecord(record);
  }

  // Synchronous commit after processing batch
  consumer.commitSync();  // Blocks until offsets written to __consumer_offsets
}

while (true) {
  ConsumerRecords<K, V> records = consumer.poll(Duration.ofMillis(100));

  for (ConsumerRecord<K, V> record : records) {
    processRecord(record);
  }

  // Asynchronous commit (non-blocking)
  consumer.commitAsync((offsets, exception) -> {
    if (exception != null) {
      log.error("Offset commit failed", exception);
    }
  });
}

// Producer side (transactional writes)
producer.initTransactions();

producer.beginTransaction();
producer.send(new ProducerRecord<>("topic", key, value));
producer.commitTransaction();  // Atomically commits messages + offsets

// Consumer side (read committed only)
props.put("isolation.level", "read_committed");

while (true) {
  ConsumerRecords<K, V> records = consumer.poll(Duration.ofMillis(100));

  for (ConsumerRecord<K, V> record : records) {
    processRecord(record);

    // Send output transactionally
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("output-topic", ...));
    producer.sendOffsetsToTransaction(
      Map.of(new TopicPartition(record.topic(), record.partition()),
             new OffsetAndMetadata(record.offset() + 1)),
      consumer.groupMetadata()
    );
    producer.commitTransaction();
  }
}

private class HeartbeatThread extends BaseHeartbeatThread {
    @Override
    protected void onSend() {
        synchronized (AbstractCoordinator.this) {
            if (!isRunning() || state != MemberState.STABLE) {
                // The coordinator may have failed while sending heartbeat
                return;
            }

            if (coordinatorUnknown()) {
                // Try to find the coordinator
                lookupCoordinator();
            } else {
                // Send heartbeat
                long now = time.milliseconds();
                lastHeartbeatSend = now;

                HeartbeatRequest.Builder requestBuilder = new HeartbeatRequest.Builder(
                    new HeartbeatRequestData()
                        .setGroupId(rebalanceConfig.groupId)
                        .setMemberId(memberId)
                        .setGenerationId(generation.generationId)
                        .setGroupInstanceId(rebalanceConfig.groupInstanceId.orElse(null))
                );

                client.send(coordinator, requestBuilder)
                    .compose(new HeartbeatResponseHandler(generation));
            }
        }
    }
}

public CoordinatorResult<HeartbeatResponseData, CoordinatorRecord> classicGroupHeartbeat(
    AuthorizableRequestContext context,
    HeartbeatRequestData request
) {
    Group group;
    try {
        group = group(request.groupId());
    } catch (GroupIdNotFoundException e) {
        throw new UnknownMemberIdException(
            String.format("Group %s not found.", request.groupId())
        );
    }

    if (group.type() == CLASSIC) {
        return classicGroupHeartbeatToClassicGroup((ClassicGroup) group, context, request);
    } else if (group.type() == CONSUMER) {
        return classicGroupHeartbeatToConsumerGroup((ConsumerGroup) group, context, request);
    } else {
        throw new UnknownMemberIdException(
            String.format("Group %s not found.", request.groupId())
        );
    }
}