@Override
    public ApiResult<CoordinatorKey, ProducerIdAndEpoch> handleSingleResponse(
        Node broker,
        CoordinatorKey key,
        AbstractResponse abstractResponse
    ) {
        InitProducerIdResponse response = (InitProducerIdResponse) abstractResponse;

        Errors error = Errors.forCode(response.data().errorCode());
        if (error != Errors.NONE) {
            return handleError(key, error);
        }

        Map<CoordinatorKey, ProducerIdAndEpoch> completed = Collections.singletonMap(key, new ProducerIdAndEpoch(
            response.data().producerId(),
            response.data().producerEpoch()
        ));

        return new ApiResult<>(completed, Collections.emptyMap(), Collections.emptyList());
    }
@Test
    public void testHandleSuccessfulResponse() {
        String transactionalId = "foo";
        CoordinatorKey key = CoordinatorKey.byTransactionalId(transactionalId);

        FenceProducersHandler handler = new FenceProducersHandler(options, logContext, requestTimeoutMs);

        short epoch = 57;
        long producerId = 7;
        InitProducerIdResponse response = new InitProducerIdResponse(new InitProducerIdResponseData()
            .setProducerEpoch(epoch)
            .setProducerId(producerId));

        ApiResult<CoordinatorKey, ProducerIdAndEpoch> result = handler.handleSingleResponse(
            node, key, response);

        assertEquals(emptyList(), result.unmappedKeys);
        assertEquals(emptyMap(), result.failedKeys);
        assertEquals(singleton(key), result.completedKeys.keySet());

        ProducerIdAndEpoch expected = new ProducerIdAndEpoch(producerId, epoch);
        assertEquals(expected, result.completedKeys.get(key));
    }