class KarelDbCoordinator extends AbstractCoordinator implements Closeable {

    @Override
    public JoinGroupRequestData.JoinGroupRequestProtocolCollection metadata() {
        ByteBuffer metadata = KarelDbProtocol.serializeMetadata(identity);
        return new JoinGroupRequestData.JoinGroupRequestProtocolCollection(
            Collections.singletonList(new JoinGroupRequestData.JoinGroupRequestProtocol()
                .setName(KDB_SUBPROTOCOL_V0)
                .setMetadata(metadata.array())).iterator());
    }

}

class KarelDbCoordinatorTest {

    private static final String LEADER_ID;
    private static final String MEMBER_ID;
    private static final String LEADER_HOST;
    private static final int LEADER_PORT;
    private static final KarelDbIdentity LEADER_INFO;
    private static final KarelDbIdentity INELIGIBLE_LEADER_INFO;
    private String groupId;
    private int sessionTimeoutMs;
    private int rebalanceTimeoutMs;
    private int heartbeatIntervalMs;
    private long retryBackoffMs;
    private MockTime time;
    private MockClient client;
    private Cluster cluster;
    private Node node;
    private Metadata metadata;
    private Metrics metrics;
    private ConsumerNetworkClient consumerClient;
    private MockRebalanceListener rebalanceListener;
    private KarelDbCoordinator coordinator;

    @Test
    public void testMetadata() {

        JoinGroupRequestData.JoinGroupRequestProtocolCollection serialized = coordinator.metadata();
        assertEquals(1, serialized.size());

        JoinGroupRequestData.JoinGroupRequestProtocol defaultMetadata = serialized.iterator().next();
        assertEquals(KarelDbCoordinator.KDB_SUBPROTOCOL_V0, defaultMetadata.name());
        KarelDbIdentity state
            = KarelDbProtocol.deserializeMetadata(ByteBuffer.wrap(defaultMetadata.metadata()));
        assertEquals(LEADER_INFO, state);
    }

}