public ByteBuffer buildResponseEnvelopePayload(AbstractResponse body) {
        return body.serializeWithHeader(header.toResponseHeader(), apiVersion());
    }
@Test
    public void testEnvelopeResponseSerde() throws Exception {
        CreateTopicsResponseData.CreatableTopicResultCollection collection =
            new CreateTopicsResponseData.CreatableTopicResultCollection();
        collection.add(new CreateTopicsResponseData.CreatableTopicResult()
            .setTopicConfigErrorCode(Errors.CLUSTER_AUTHORIZATION_FAILED.code())
            .setNumPartitions(5));
        CreateTopicsResponseData expectedResponse = new CreateTopicsResponseData()
            .setThrottleTimeMs(10)
            .setTopics(collection);

        int correlationId = 15;
        String clientId = "clientId";
        RequestHeader header = new RequestHeader(ApiKeys.CREATE_TOPICS, ApiKeys.CREATE_TOPICS.latestVersion(),
            clientId, correlationId);

        RequestContext context = new RequestContext(header, "0", InetAddress.getLocalHost(),
            KafkaPrincipal.ANONYMOUS, new ListenerName("ssl"), SecurityProtocol.SASL_SSL,
            ClientInformation.EMPTY, true);

        ByteBuffer buffer = context.buildResponseEnvelopePayload(new CreateTopicsResponse(expectedResponse));
        assertEquals(buffer.capacity(), buffer.limit(), "Buffer limit and capacity should be the same");
        CreateTopicsResponse parsedResponse = (CreateTopicsResponse) AbstractResponse.parseResponse(buffer, header);
        assertEquals(expectedResponse, parsedResponse.data());
    }