/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.pulsar.broker.admin.impl;

import static javax.ws.rs.core.Response.Status.INTERNAL_SERVER_ERROR;
import static javax.ws.rs.core.Response.Status.METHOD_NOT_ALLOWED;
import static javax.ws.rs.core.Response.Status.NOT_FOUND;
import static javax.ws.rs.core.Response.Status.SERVICE_UNAVAILABLE;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import javax.ws.rs.container.AsyncResponse;
import javax.ws.rs.core.Response;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.admin.AdminResource;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor;
import org.apache.pulsar.broker.web.RestException;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.Transactions;
import org.apache.pulsar.client.api.transaction.TxnID;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.SystemTopicNames;
import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.policies.data.SnapshotSystemTopicInternalStats;
import org.apache.pulsar.common.policies.data.TransactionBufferInternalStats;
import org.apache.pulsar.common.policies.data.TransactionBufferStats;
import org.apache.pulsar.common.policies.data.TransactionCoordinatorInfo;
import org.apache.pulsar.common.policies.data.TransactionCoordinatorInternalStats;
import org.apache.pulsar.common.policies.data.TransactionCoordinatorStats;
import org.apache.pulsar.common.policies.data.TransactionInBufferStats;
import org.apache.pulsar.common.policies.data.TransactionInPendingAckStats;
import org.apache.pulsar.common.policies.data.TransactionLogStats;
import org.apache.pulsar.common.policies.data.TransactionMetadata;
import org.apache.pulsar.common.policies.data.TransactionPendingAckInternalStats;
import org.apache.pulsar.common.policies.data.TransactionPendingAckStats;
import org.apache.pulsar.common.stats.PositionInPendingAckStats;
import org.apache.pulsar.common.util.Codec;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.transaction.coordinator.TransactionCoordinatorID;
import org.apache.pulsar.transaction.coordinator.TransactionMetadataStore;
import org.apache.pulsar.transaction.coordinator.TxnMeta;
import org.apache.pulsar.transaction.coordinator.exceptions.CoordinatorException.CoordinatorNotFoundException;
import org.apache.pulsar.transaction.coordinator.exceptions.CoordinatorException.TransactionNotFoundException;
import org.apache.pulsar.transaction.coordinator.impl.MLTransactionMetadataStore;

@Slf4j
public abstract class TransactionsBase extends AdminResource {

    protected void internalListCoordinators(AsyncResponse asyncResponse) {
        final PulsarAdmin admin;
        try {
            admin = pulsar().getAdminClient();
        } catch (PulsarServerException ex) {
            asyncResponse.resume(new RestException(ex));
            return;
        }
        Map<Integer, TransactionCoordinatorInfo> result = new HashMap<>();
        admin.lookups()
                .lookupPartitionedTopicAsync(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN.getPartitionedTopicName())
                .thenAccept(map -> {
                    map.forEach((topicPartition, brokerServiceUrl) -> {
                        final int coordinatorId = TopicName.getPartitionIndex(topicPartition);
                        result.put(coordinatorId, new TransactionCoordinatorInfo(coordinatorId, brokerServiceUrl));
                    });

                    asyncResponse.resume(result.values());
                })
                .exceptionally(ex -> {
                    log.error("[{}] Failed to list transaction coordinators: {}",
                            clientAppId(), ex.getMessage(), ex);
                    resumeAsyncResponseExceptionally(asyncResponse, ex);
                    return null;
                });
    }

    protected void internalGetCoordinatorStats(AsyncResponse asyncResponse, boolean authoritative,
                                               Integer coordinatorId) {
        if (coordinatorId != null) {
            validateTopicOwnership(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN.getPartition(coordinatorId),
                    authoritative);
            TransactionMetadataStore transactionMetadataStore =
                    pulsar().getTransactionMetadataStoreService().getStores()
                            .get(TransactionCoordinatorID.get(coordinatorId));
            if (transactionMetadataStore == null) {
                asyncResponse.resume(new RestException(NOT_FOUND,
                        "Transaction coordinator not found! coordinator id : " + coordinatorId));
                return;
            }
            asyncResponse.resume(transactionMetadataStore.getCoordinatorStats());
        } else {
            getPartitionedTopicMetadataAsync(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN,
                    false, false).thenAccept(partitionMetadata -> {
                if (partitionMetadata.partitions == 0) {
                    asyncResponse.resume(new RestException(Response.Status.NOT_FOUND,
                            "Transaction coordinator not found"));
                    return;
                }
                List<CompletableFuture<TransactionCoordinatorStats>> transactionMetadataStoreInfoFutures =
                        new ArrayList<>();
                for (int i = 0; i < partitionMetadata.partitions; i++) {
                    try {
                        transactionMetadataStoreInfoFutures
                                .add(pulsar().getAdminClient().transactions().getCoordinatorStatsByIdAsync(i));
                    } catch (PulsarServerException e) {
                        asyncResponse.resume(new RestException(e));
                        return;
                    }
                }
                Map<Integer, TransactionCoordinatorStats> stats = new HashMap<>();
                FutureUtil.waitForAll(transactionMetadataStoreInfoFutures).whenComplete((result, e) -> {
                    if (e != null) {
                        asyncResponse.resume(new RestException(e));
                        return;
                    }

                    for (int i = 0; i < transactionMetadataStoreInfoFutures.size(); i++) {
                        try {
                            stats.put(i, transactionMetadataStoreInfoFutures.get(i).get());
                        } catch (Exception exception) {
                            asyncResponse.resume(new RestException(exception.getCause()));
                            return;
                        }
                    }

                    asyncResponse.resume(stats);
                });
            }).exceptionally(ex -> {
                log.error("[{}] Failed to get transaction coordinator state.", clientAppId(), ex);
                resumeAsyncResponseExceptionally(asyncResponse, ex);
                return null;
            });
        }
    }

    protected CompletableFuture<TransactionInPendingAckStats> internalGetTransactionInPendingAckStats(
            boolean authoritative, long mostSigBits, long leastSigBits, String subName) {
        return getExistingPersistentTopicAsync(authoritative)
                .thenApply(topic -> topic.getTransactionInPendingAckStats(new TxnID(mostSigBits, leastSigBits),
                        subName));
    }

    protected CompletableFuture<TransactionInBufferStats> internalGetTransactionInBufferStats(
            boolean authoritative, long mostSigBits, long leastSigBits) {
        return getExistingPersistentTopicAsync(authoritative)
                .thenApply(topic -> topic.getTransactionInBufferStats(new TxnID(mostSigBits, leastSigBits)));
    }

    protected CompletableFuture<TransactionBufferStats> internalGetTransactionBufferStats(boolean authoritative,
                                                                                          boolean lowWaterMarks,
                                                                                          boolean segmentStats) {
        return getExistingPersistentTopicAsync(authoritative)
                .thenApply(topic -> topic.getTransactionBufferStats(lowWaterMarks, segmentStats));
    }

    protected CompletableFuture<TransactionPendingAckStats> internalGetPendingAckStats(
            boolean authoritative, String subName, boolean lowWaterMarks) {
        return getExistingPersistentTopicAsync(authoritative)
                .thenApply(topic -> topic.getTransactionPendingAckStats(subName, lowWaterMarks));
    }

    protected void internalGetTransactionMetadata(AsyncResponse asyncResponse,
                                                  boolean authoritative, int mostSigBits, long leastSigBits) {
        try {
            validateTopicOwnership(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN.getPartition(mostSigBits),
                    authoritative);
            CompletableFuture<TransactionMetadata> transactionMetadataFuture = new CompletableFuture<>();
            TxnMeta txnMeta = pulsar().getTransactionMetadataStoreService()
                    .getTxnMeta(new TxnID(mostSigBits, leastSigBits)).get();
            getTransactionMetadata(txnMeta, transactionMetadataFuture);
            asyncResponse.resume(transactionMetadataFuture.get(10, TimeUnit.SECONDS));
        } catch (Exception e) {
            if (e instanceof ExecutionException) {
                if (e.getCause() instanceof CoordinatorNotFoundException
                        || e.getCause() instanceof TransactionNotFoundException) {
                    asyncResponse.resume(new RestException(NOT_FOUND, e.getCause()));
                    return;
                }
                asyncResponse.resume(new RestException(e.getCause()));
            } else {
                asyncResponse.resume(new RestException(e));
            }
        }
    }

    private void getTransactionMetadata(TxnMeta txnMeta,
                                        CompletableFuture<TransactionMetadata> transactionMetadataFuture)
            throws PulsarServerException {
        Transactions transactions = pulsar().getAdminClient().transactions();
        TransactionMetadata transactionMetadata = new TransactionMetadata();
        TxnID txnID = txnMeta.id();
        transactionMetadata.txnId = txnID.toString();
        transactionMetadata.status = txnMeta.status().name();
        transactionMetadata.openTimestamp = txnMeta.getOpenTimestamp();
        transactionMetadata.timeoutAt = txnMeta.getTimeoutAt();
        transactionMetadata.owner = txnMeta.getOwner();

        List<CompletableFuture<TransactionInPendingAckStats>> ackedPartitionsFutures = new ArrayList<>();
        Map<String, Map<String, CompletableFuture<TransactionInPendingAckStats>>> ackFutures = new HashMap<>();
        txnMeta.ackedPartitions().forEach(transactionSubscription -> {
            String topic = transactionSubscription.getTopic();
            String subName = transactionSubscription.getSubscription();
            CompletableFuture<TransactionInPendingAckStats> future =
                    transactions.getTransactionInPendingAckStatsAsync(txnID, topic, subName);
            ackedPartitionsFutures.add(future);
            if (ackFutures.containsKey(topic)) {
                ackFutures.get(topic)
                        .put(transactionSubscription.getSubscription(), future);
            } else {
                Map<String, CompletableFuture<TransactionInPendingAckStats>> pendingAckStatsMap =
                        new HashMap<>();
                pendingAckStatsMap.put(transactionSubscription.getSubscription(), future);
                ackFutures.put(topic, pendingAckStatsMap);
            }
        });

        List<CompletableFuture<TransactionInBufferStats>> producedPartitionsFutures = new ArrayList<>();
        Map<String, CompletableFuture<TransactionInBufferStats>> produceFutures = new HashMap<>();
        txnMeta.producedPartitions().forEach(topic -> {
            CompletableFuture<TransactionInBufferStats> future =
                    transactions.getTransactionInBufferStatsAsync(txnID, topic);
            producedPartitionsFutures.add(future);
            produceFutures.put(topic, future);

        });

        FutureUtil.waitForAll(ackedPartitionsFutures).whenComplete((v, e) -> {
            if (e != null) {
                transactionMetadataFuture.completeExceptionally(e);
                return;
            }

            FutureUtil.waitForAll(producedPartitionsFutures).whenComplete((x, t) -> {
                if (t != null) {
                    transactionMetadataFuture.completeExceptionally(e);
                    return;
                }

                Map<String, Map<String, TransactionInPendingAckStats>> ackedPartitions = new HashMap<>();
                Map<String, TransactionInBufferStats> producedPartitions = new HashMap<>();

                for (String topic : ackFutures.keySet()) {
                    Map<String, TransactionInPendingAckStats> subs = new HashMap<>();
                    for (String sub : ackFutures.get(topic).keySet()) {
                        try {
                            subs.put(sub, ackFutures.get(topic).get(sub).get());
                        } catch (Exception exception) {
                            transactionMetadataFuture.completeExceptionally(exception);
                            return;
                        }
                    }

                    ackedPartitions.put(topic, subs);
                }

                for (String topic : produceFutures.keySet()) {
                    try {
                        producedPartitions.put(topic, produceFutures.get(topic).get());
                    } catch (Exception exception) {
                        transactionMetadataFuture.completeExceptionally(exception);
                        return;
                    }
                }
                transactionMetadata.ackedPartitions = ackedPartitions;
                transactionMetadata.producedPartitions = producedPartitions;
                transactionMetadataFuture.complete(transactionMetadata);
            });
        });
    }

    protected void internalGetSlowTransactions(AsyncResponse asyncResponse,
                                               boolean authoritative, long timeout, Integer coordinatorId) {
        try {
            if (coordinatorId != null) {
                validateTopicOwnership(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN.getPartition(coordinatorId),
                        authoritative);
                TransactionMetadataStore transactionMetadataStore =
                        pulsar().getTransactionMetadataStoreService().getStores()
                                .get(TransactionCoordinatorID.get(coordinatorId));
                if (transactionMetadataStore == null) {
                    asyncResponse.resume(new RestException(NOT_FOUND,
                            "Transaction coordinator not found! coordinator id : " + coordinatorId));
                    return;
                }
                List<TxnMeta> transactions = transactionMetadataStore.getSlowTransactions(timeout);
                List<CompletableFuture<TransactionMetadata>> completableFutures = new ArrayList<>();
                for (TxnMeta txnMeta : transactions) {
                    CompletableFuture<TransactionMetadata> completableFuture = new CompletableFuture<>();
                    getTransactionMetadata(txnMeta, completableFuture);
                    completableFutures.add(completableFuture);
                }

                FutureUtil.waitForAll(completableFutures).whenComplete((v, e) -> {
                    if (e != null) {
                        asyncResponse.resume(new RestException(e.getCause()));
                        return;
                    }

                    Map<String, TransactionMetadata> transactionMetadata = new HashMap<>();
                    for (CompletableFuture<TransactionMetadata> future : completableFutures) {
                        try {
                            transactionMetadata.put(future.get().txnId, future.get());
                        } catch (Exception exception) {
                            asyncResponse.resume(new RestException(exception.getCause()));
                            return;
                        }
                    }
                    asyncResponse.resume(transactionMetadata);
                });
            } else {
                getPartitionedTopicMetadataAsync(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN,
                        false, false).thenAccept(partitionMetadata -> {
                    if (partitionMetadata.partitions == 0) {
                        asyncResponse.resume(new RestException(Response.Status.NOT_FOUND,
                                "Transaction coordinator not found"));
                        return;
                    }
                    List<CompletableFuture<Map<String, TransactionMetadata>>> completableFutures =
                            new ArrayList<>();
                    for (int i = 0; i < partitionMetadata.partitions; i++) {
                        try {
                            completableFutures
                                    .add(pulsar().getAdminClient().transactions()
                                            .getSlowTransactionsByCoordinatorIdAsync(i, timeout,
                                                    TimeUnit.MILLISECONDS));
                        } catch (PulsarServerException e) {
                            asyncResponse.resume(new RestException(e));
                            return;
                        }
                    }
                    Map<String, TransactionMetadata> transactionMetadataMaps = new HashMap<>();
                    FutureUtil.waitForAll(completableFutures).whenComplete((result, e) -> {
                        if (e != null) {
                            asyncResponse.resume(new RestException(e));
                            return;
                        }

                        for (CompletableFuture<Map<String, TransactionMetadata>> transactionMetadataMap
                                : completableFutures) {
                            try {
                                transactionMetadataMaps.putAll(transactionMetadataMap.get());
                            } catch (Exception exception) {
                                asyncResponse.resume(new RestException(exception.getCause()));
                                return;
                            }
                        }
                        asyncResponse.resume(transactionMetadataMaps);
                    });
                }).exceptionally(ex -> {
                    log.error("[{}] Failed to get transaction coordinator state.", clientAppId(), ex);
                    resumeAsyncResponseExceptionally(asyncResponse, ex);
                    return null;
                });

            }
        } catch (Exception e) {
            asyncResponse.resume(new RestException(e));
        }
    }

    protected void internalGetCoordinatorInternalStats(AsyncResponse asyncResponse, boolean authoritative,
                                                       boolean metadata, int coordinatorId) {
        try {
            TopicName topicName = SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN.getPartition(coordinatorId);
            validateTopicOwnership(topicName, authoritative);
            TransactionMetadataStore metadataStore = pulsar().getTransactionMetadataStoreService()
                    .getStores().get(TransactionCoordinatorID.get(coordinatorId));
            if (metadataStore == null) {
                asyncResponse.resume(new RestException(NOT_FOUND,
                        "Transaction coordinator not found! coordinator id : " + coordinatorId));
                return;
            }
            if (metadataStore instanceof MLTransactionMetadataStore) {
                ManagedLedger managedLedger = ((MLTransactionMetadataStore) metadataStore).getManagedLedger();
                TransactionCoordinatorInternalStats transactionCoordinatorInternalStats =
                        new TransactionCoordinatorInternalStats();
                TransactionLogStats transactionLogStats = new TransactionLogStats();
                transactionLogStats.managedLedgerName = managedLedger.getName();
                transactionLogStats.managedLedgerInternalStats =
                        managedLedger.getManagedLedgerInternalStats(metadata).get();
                transactionCoordinatorInternalStats.transactionLogStats = transactionLogStats;
                asyncResponse.resume(transactionCoordinatorInternalStats);
            } else {
                asyncResponse.resume(new RestException(METHOD_NOT_ALLOWED,
                        "Broker don't use MLTransactionMetadataStore!"));
            }
        } catch (Exception e) {
            resumeAsyncResponseExceptionally(asyncResponse, e);
        }
    }

    protected CompletableFuture<TransactionPendingAckInternalStats> internalGetPendingAckInternalStats(
            boolean authoritative, String subName, boolean metadata) {
        return getExistingPersistentTopicAsync(authoritative)
                .thenCompose(topic -> topic.getPendingAckManagedLedger(subName))
                .thenCompose(managedLedger ->
                        managedLedger.getManagedLedgerInternalStats(metadata)
                            .thenApply(internalStats -> {
                                TransactionLogStats pendingAckLogStats = new TransactionLogStats();
                                pendingAckLogStats.managedLedgerName = managedLedger.getName();
                                pendingAckLogStats.managedLedgerInternalStats = internalStats;
                                return pendingAckLogStats;
                            })
                            .thenApply(pendingAckLogStats -> {
                                TransactionPendingAckInternalStats stats = new TransactionPendingAckInternalStats();
                                stats.pendingAckLogStats = pendingAckLogStats;
                                return stats;
                            })
                );
    }

    protected CompletableFuture<TransactionBufferInternalStats> internalGetTransactionBufferInternalStats(
            boolean authoritative, boolean metadata) {
        TransactionBufferInternalStats transactionBufferInternalStats = new TransactionBufferInternalStats();
        return getExistingPersistentTopicAsync(authoritative)
                .thenCompose(topic -> {
                    AbortedTxnProcessor.SnapshotType snapshotType = topic.getTransactionBuffer().getSnapshotType();
                    if (snapshotType == null) {
                        return FutureUtil.failedFuture(new RestException(NOT_FOUND,
                                "Transaction buffer Snapshot for the topic does not exist"));
                    } else if (snapshotType == AbortedTxnProcessor.SnapshotType.Segment) {
                        transactionBufferInternalStats.snapshotType = snapshotType.toString();
                        TopicName segmentTopic = TopicName.get(TopicDomain.persistent.toString(), namespaceName,
                                SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT_SEGMENTS);
                        CompletableFuture<SnapshotSystemTopicInternalStats> segmentInternalStatsFuture =
                                getTxnSnapshotInternalStats(segmentTopic, metadata);
                        TopicName indexTopic = TopicName.get(TopicDomain.persistent.toString(),
                                namespaceName,
                                SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT_INDEXES);
                        CompletableFuture<SnapshotSystemTopicInternalStats> segmentIndexInternalStatsFuture =
                                getTxnSnapshotInternalStats(indexTopic, metadata);
                        return segmentIndexInternalStatsFuture
                                .thenCombine(segmentInternalStatsFuture, (indexStats, segmentStats) -> {
                                    transactionBufferInternalStats.segmentIndexInternalStats = indexStats;
                                    transactionBufferInternalStats.segmentInternalStats = segmentStats;
                                    return transactionBufferInternalStats;
                                });
                    } else if (snapshotType == AbortedTxnProcessor.SnapshotType.Single) {
                        transactionBufferInternalStats.snapshotType = snapshotType.toString();
                        TopicName singleSnapshotTopic = TopicName.get(TopicDomain.persistent.toString(), namespaceName,
                                SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT);
                        return getTxnSnapshotInternalStats(singleSnapshotTopic, metadata)
                                .thenApply(snapshotSystemTopicInternalStats -> {
                                   transactionBufferInternalStats.singleSnapshotSystemTopicInternalStats =
                                           snapshotSystemTopicInternalStats;
                                   return transactionBufferInternalStats;
                                });
                    }
                    return FutureUtil.failedFuture(new RestException(INTERNAL_SERVER_ERROR, "Unknown SnapshotType "
                            + snapshotType));
                });
    }

    private CompletableFuture<SnapshotSystemTopicInternalStats> getTxnSnapshotInternalStats(TopicName topicName,
                                                                                            boolean metadata) {
        final PulsarAdmin admin;
        try {
            admin = pulsar().getAdminClient();
        } catch (PulsarServerException e) {
            return FutureUtil.failedFuture(new RestException(e));
        }
        return admin.topics().getInternalStatsAsync(topicName.toString(), metadata)
                        .thenApply(persistentTopicInternalStats -> {
                            SnapshotSystemTopicInternalStats
                                    snapshotSystemTopicInternalStats = new SnapshotSystemTopicInternalStats();
                            snapshotSystemTopicInternalStats.managedLedgerInternalStats = persistentTopicInternalStats;
                            snapshotSystemTopicInternalStats.managedLedgerName = topicName.getEncodedLocalName();
                            return snapshotSystemTopicInternalStats;
                        });
    }

    protected CompletableFuture<PersistentTopic> getExistingPersistentTopicAsync(boolean authoritative) {
        return validateTopicOwnershipAsync(topicName, authoritative).thenCompose(__ -> {
            CompletableFuture<Optional<Topic>> topicFuture = pulsar().getBrokerService()
                    .getTopics().get(topicName.toString());
            if (topicFuture == null) {
                return FutureUtil.failedFuture(new RestException(NOT_FOUND,
                        String.format("Topic not found %s", topicName.toString())));
            }
            return topicFuture.thenCompose(optionalTopic -> {
                if (!optionalTopic.isPresent()) {
                    return FutureUtil.failedFuture(new RestException(NOT_FOUND,
                            String.format("Topic not found %s", topicName.toString())));
                }
                return CompletableFuture.completedFuture((PersistentTopic) optionalTopic.get());
            });
        });
    }

    protected void checkTransactionCoordinatorEnabled() {
        if (!pulsar().getConfig().isTransactionCoordinatorEnabled()) {
           throw new RestException(SERVICE_UNAVAILABLE,
                    "This Broker is not configured with transactionCoordinatorEnabled=true.");
        }
    }

    protected void validateTopicName(String property, String namespace, String encodedTopic) {
        String topic = Codec.decode(encodedTopic);
        try {
            this.namespaceName = NamespaceName.get(property, namespace);
            this.topicName = TopicName.get(TopicDomain.persistent.toString(), namespaceName, topic);
        } catch (IllegalArgumentException e) {
            log.warn("[{}] Failed to validate topic name {}://{}/{}/{}", clientAppId(), domain(), property, namespace,
                    topic, e);
            throw new RestException(Response.Status.PRECONDITION_FAILED, "Topic name is not valid");
        }
    }

    protected CompletableFuture<Void> internalScaleTransactionCoordinators(int replicas) {
        return validateSuperUserAccessAsync()
                .thenCompose((ignore) -> namespaceResources().getPartitionedTopicResources()
                        .updatePartitionedTopicAsync(SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN, p -> {
                            if (p.partitions >= replicas) {
                                throw new RestException(Response.Status.NOT_ACCEPTABLE,
                                        "Number of transaction coordinators should "
                                                + "be more than the current number of transaction coordinator");
                            }
                            return new PartitionedTopicMetadata(replicas);
                        }));
    }

    protected CompletableFuture<PositionInPendingAckStats> internalGetPositionStatsPendingAckStats(
            boolean authoritative, String subName, PositionImpl position, Integer batchIndex) {
        CompletableFuture<PositionInPendingAckStats> completableFuture = new CompletableFuture<>();
        getExistingPersistentTopicAsync(authoritative)
                .thenAccept(topic -> {
                    PositionInPendingAckStats result = topic.getSubscription(subName)
                    .checkPositionInPendingAckState(position, batchIndex);
                    completableFuture.complete(result);
                }).exceptionally(ex -> {
                    completableFuture.completeExceptionally(ex);
                    return null;
        });
        return completableFuture;
    }
}