static String consumerGroupJoinKey(String groupId, String memberId) {
        return "join-" + groupId + "-" + memberId;
    }
@Test
    public void testConsumerGroupMemberUsingClassicProtocolFencedWhenJoinTimeout() {
        String groupId = "group-id";
        String memberId = Uuid.randomUuid().toString();
        int rebalanceTimeout = 500;

        List<ConsumerGroupMemberMetadataValue.ClassicProtocol> protocols = Collections.singletonList(
            new ConsumerGroupMemberMetadataValue.ClassicProtocol()
                .setName("range")
                .setMetadata(Utils.toArray(ConsumerProtocol.serializeSubscription(
                    new ConsumerPartitionAssignor.Subscription(
                        Collections.singletonList("foo"),
                        null,
                        Collections.emptyList()
                    )
                )))
        );

        // Consumer group with a member using the classic protocol whose member epoch is smaller than the group epoch.
        GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
            .withConsumerGroupAssignors(Collections.singletonList(new MockPartitionAssignor("range")))
            .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
                .withMember(new ConsumerGroupMember.Builder(memberId)
                    .setRebalanceTimeoutMs(rebalanceTimeout)
                    .setClassicMemberMetadata(
                        new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
                            .setSessionTimeoutMs(5000)
                            .setSupportedProtocols(protocols)
                    )
                    .setMemberEpoch(9)
                    .build()))
            .build();

        // Heartbeat to schedule the join timeout.
        HeartbeatRequestData request = new HeartbeatRequestData()
            .setGroupId(groupId)
            .setMemberId(memberId)
            .setGenerationId(9);
        assertEquals(
            Errors.REBALANCE_IN_PROGRESS.code(),
            context.sendClassicGroupHeartbeat(request).response().errorCode()
        );
        context.assertSessionTimeout(groupId, memberId, 5000);
        context.assertJoinTimeout(groupId, memberId, rebalanceTimeout);

        // Advance clock by rebalance timeout + 1.
        List<ExpiredTimeout<Void, CoordinatorRecord>> timeouts = context.sleep(rebalanceTimeout + 1);

        // The member is fenced from the group.
        assertEquals(1, timeouts.size());
        ExpiredTimeout<Void, CoordinatorRecord> timeout = timeouts.get(0);
        assertEquals(consumerGroupJoinKey(groupId, memberId), timeout.key);
        assertRecordsEquals(
            Arrays.asList(
                // The member is removed.
                GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId, memberId),
                GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId, memberId),
                GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId, memberId),

                // The group epoch is bumped.
                GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 11)
            ),
            timeout.result.records()
        );
    }