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()); } KarelDbCoordinator(
        LogContext logContext,
        ConsumerNetworkClient client,
        String groupId,
        int rebalanceTimeoutMs,
        int sessionTimeoutMs,
        int heartbeatIntervalMs,
        Metrics metrics,
        String metricGrpPrefix,
        Time time,
        long retryBackoffMs,
        KarelDbIdentity identity,
        KarelDbRebalanceListener listener); @Override String protocolType(); void poll(long timeout); @Override JoinGroupRequestData.JoinGroupRequestProtocolCollection metadata(); static final String KDB_SUBPROTOCOL_V0; }
@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); }