@Override
    public ApiResult<TopicPartition, PartitionProducerState> handleResponse(
        Node broker,
        Set<TopicPartition> keys,
        AbstractResponse abstractResponse
    ) {
        DescribeProducersResponse response = (DescribeProducersResponse) abstractResponse;
        Map<TopicPartition, PartitionProducerState> completed = new HashMap<>();
        Map<TopicPartition, Throwable> failed = new HashMap<>();
        List<TopicPartition> unmapped = new ArrayList<>();

        for (DescribeProducersResponseData.TopicResponse topicResponse : response.data().topics()) {
            for (DescribeProducersResponseData.PartitionResponse partitionResponse : topicResponse.partitions()) {
                TopicPartition topicPartition = new TopicPartition(
                    topicResponse.name(), partitionResponse.partitionIndex());

                Errors error = Errors.forCode(partitionResponse.errorCode());
                if (error != Errors.NONE) {
                    ApiError apiError = new ApiError(error, partitionResponse.errorMessage());
                    handlePartitionError(topicPartition, apiError, failed, unmapped);
                    continue;
                }

                List<ProducerState> activeProducers = partitionResponse.activeProducers().stream()
                    .map(activeProducer -> {
                        OptionalLong currentTransactionFirstOffset =
                            activeProducer.currentTxnStartOffset() < 0 ?
                                OptionalLong.empty() :
                                OptionalLong.of(activeProducer.currentTxnStartOffset());
                        OptionalInt coordinatorEpoch =
                            activeProducer.coordinatorEpoch() < 0 ?
                                OptionalInt.empty() :
                                OptionalInt.of(activeProducer.coordinatorEpoch());

                        return new ProducerState(
                            activeProducer.producerId(),
                            activeProducer.producerEpoch(),
                            activeProducer.lastSequence(),
                            activeProducer.lastTimestamp(),
                            coordinatorEpoch,
                            currentTransactionFirstOffset
                        );
                    }).collect(Collectors.toList());

                completed.put(topicPartition, new PartitionProducerState(activeProducers));
            }
        }
        return new ApiResult<>(completed, failed, unmapped);
    }
@Test
    public void testCompletedResult() {
        TopicPartition topicPartition = new TopicPartition("foo", 5);
        DescribeProducersOptions options = new DescribeProducersOptions().brokerId(1);
        DescribeProducersHandler handler = newHandler(options);

        PartitionResponse partitionResponse = sampleProducerState(topicPartition);
        DescribeProducersResponse response = describeProducersResponse(
            singletonMap(topicPartition, partitionResponse)
        );
        Node node = new Node(3, "host", 1);

        ApiResult<TopicPartition, PartitionProducerState> result =
            handler.handleResponse(node, mkSet(topicPartition), response);

        assertEquals(mkSet(topicPartition), result.completedKeys.keySet());
        assertEquals(emptyMap(), result.failedKeys);
        assertEquals(emptyList(), result.unmappedKeys);

        PartitionProducerState producerState = result.completedKeys.get(topicPartition);
        assertMatchingProducers(partitionResponse, producerState);
    }