public RequestAndSize parseRequest(ByteBuffer buffer) {
        if (isUnsupportedApiVersionsRequest()) {
            // Unsupported ApiVersion requests are treated as v0 requests and are not parsed
            ApiVersionsRequest apiVersionsRequest = new ApiVersionsRequest(new ApiVersionsRequestData(), (short) 0, header.apiVersion());
            return new RequestAndSize(apiVersionsRequest, 0);
        } else {
            ApiKeys apiKey = header.apiKey();
            try {
                short apiVersion = header.apiVersion();
                return AbstractRequest.parseRequest(apiKey, apiVersion, buffer);
            } catch (Throwable ex) {
                throw new InvalidRequestException("Error getting request for apiKey: " + apiKey +
                        ", apiVersion: " + header.apiVersion() +
                        ", connectionId: " + connectionId +
                        ", listenerName: " + listenerName +
                        ", principal: " + principal, ex);
            }
        }
    }
@Test
    public void testInvalidRequestForArrayList() throws UnknownHostException {
        short version = (short) 5; // choose a version with fixed length encoding, for simplicity
        ByteBuffer corruptBuffer = produceRequest(version);
        // corrupt the length of the partitions array
        corruptBuffer.putInt(17, Integer.MAX_VALUE);

        RequestHeader header = new RequestHeader(ApiKeys.PRODUCE, version, "console-producer", 3);
        RequestContext context = new RequestContext(header, "0", InetAddress.getLocalHost(),
                KafkaPrincipal.ANONYMOUS, new ListenerName("ssl"), SecurityProtocol.SASL_SSL,
                ClientInformation.EMPTY, true);

        String msg = assertThrows(InvalidRequestException.class,
                () -> context.parseRequest(corruptBuffer)).getCause().getMessage();
        assertEquals(
                "Tried to allocate a collection of size 2147483647, but there are only 8 bytes remaining.", msg);
    }