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

        for (DescribeTransactionsResponseData.TransactionState transactionState : response.data().transactionStates()) {
            CoordinatorKey transactionalIdKey = CoordinatorKey.byTransactionalId(
                transactionState.transactionalId());
            if (!keys.contains(transactionalIdKey)) {
                log.warn("Response included transactionalId `{}`, which was not requested",
                    transactionState.transactionalId());
                continue;
            }

            Errors error = Errors.forCode(transactionState.errorCode());
            if (error != Errors.NONE) {
                handleError(transactionalIdKey, error, failed, unmapped);
                continue;
            }

            OptionalLong transactionStartTimeMs = transactionState.transactionStartTimeMs() < 0 ?
                OptionalLong.empty() :
                OptionalLong.of(transactionState.transactionStartTimeMs());

            completed.put(transactionalIdKey, new TransactionDescription(
                broker.id(),
                TransactionState.parse(transactionState.transactionState()),
                transactionState.producerId(),
                transactionState.producerEpoch(),
                transactionState.transactionTimeoutMs(),
                transactionStartTimeMs,
                collectTopicPartitions(transactionState)
            ));
        }

        return new ApiResult<>(completed, failed, unmapped);
    }
@Test
    public void testHandleSuccessfulResponse() {
        String transactionalId1 = "foo";
        String transactionalId2 = "bar";

        Set<String> transactionalIds = mkSet(transactionalId1, transactionalId2);
        DescribeTransactionsHandler handler = new DescribeTransactionsHandler(logContext);

        DescribeTransactionsResponseData.TransactionState transactionState1 =
            sampleTransactionState1(transactionalId1);
        DescribeTransactionsResponseData.TransactionState transactionState2 =
            sampleTransactionState2(transactionalId2);

        Set<CoordinatorKey> keys = coordinatorKeys(transactionalIds);
        DescribeTransactionsResponse response = new DescribeTransactionsResponse(new DescribeTransactionsResponseData()
            .setTransactionStates(asList(transactionState1, transactionState2)));

        ApiResult<CoordinatorKey, TransactionDescription> result = handler.handleResponse(
            node, keys, response);

        assertEquals(keys, result.completedKeys.keySet());
        assertMatchingTransactionState(node.id(), transactionState1,
            result.completedKeys.get(CoordinatorKey.byTransactionalId(transactionalId1)));
        assertMatchingTransactionState(node.id(), transactionState2,
            result.completedKeys.get(CoordinatorKey.byTransactionalId(transactionalId2)));
    }