class WorkerCoordinator extends AbstractCoordinator implements Closeable {

    @Override
    public JoinGroupRequestProtocolCollection metadata() {
        configSnapshot = configStorage.snapshot();
        final ExtendedAssignment localAssignmentSnapshot = assignmentSnapshot;
        ExtendedWorkerState workerState = new ExtendedWorkerState(restUrl, configSnapshot.offset(), localAssignmentSnapshot);
        switch (protocolCompatibility) {
            case EAGER:
                return ConnectProtocol.metadataRequest(workerState);
            case COMPATIBLE:
                return IncrementalCooperativeConnectProtocol.metadataRequest(workerState, false);
            case SESSIONED:
                return IncrementalCooperativeConnectProtocol.metadataRequest(workerState, true);
            default:
                throw new IllegalStateException("Unknown Connect protocol compatibility mode " + protocolCompatibility);
        }
    }

    public  WorkerCoordinator(GroupRebalanceConfig config,
                             LogContext logContext,
                             ConsumerNetworkClient client,
                             Metrics metrics,
                             String metricGrpPrefix,
                             Time time,
                             String restUrl,
                             ConfigBackingStore configStorage,
                             WorkerRebalanceListener listener,
                             ConnectProtocolCompatibility protocolCompatibility,
                             int maxDelay);

}