static Cluster parseDescribeClusterResponse(DescribeClusterResponseData response) {
        ApiError apiError = new ApiError(response.errorCode(), response.errorMessage());
        if (apiError.isFailure()) {
            throw apiError.exception();
        }
        if (response.endpointType() != EndpointType.CONTROLLER.id()) {
            throw new MismatchedEndpointTypeException("Expected response from CONTROLLER " +
                "endpoint, but got response from endpoint type " + (int) response.endpointType());
        }
        List<Node> nodes = new ArrayList<>();
        Node controllerNode = null;
        for (DescribeClusterResponseData.DescribeClusterBroker node : response.brokers()) {
            Node newNode = new Node(node.brokerId(), node.host(), node.port(), node.rack());
            nodes.add(newNode);
            if (node.brokerId() == response.controllerId()) {
                controllerNode = newNode;
            }
        }
        return new Cluster(response.clusterId(),
            nodes,
            Collections.emptyList(),
            Collections.emptySet(),
            Collections.emptySet(),
            controllerNode);
    }
@Test
    public void testParseDescribeClusterResponseWithError() {
        assertThrows(MismatchedEndpointTypeException.class,
            () -> KafkaAdminClient.parseDescribeClusterResponse(new DescribeClusterResponseData().
                setErrorCode(Errors.MISMATCHED_ENDPOINT_TYPE.code()).
                setErrorMessage("The request was sent to an endpoint of type BROKER, " +
                        "but we wanted an endpoint of type CONTROLLER")));
    }