text:
package org.apache.kafka.clients.producer.internals;

import org.apache.kafka.clients.ClientResponse;
import org.apache.kafka.clients.RequestCompletionHandler;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.AuthenticationException;
import org.apache.kafka.common.errors.GroupAuthorizationException;
import org.apache.kafka.common.errors.TopicAuthorizationException;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.record.RecordBatch;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.AbstractResponse;
import org.apache.kafka.common.requests.AddOffsetsToTxnRequest;
import org.apache.kafka.common.requests.AddOffsetsToTxnResponse;
import org.apache.kafka.common.requests.AddPartitionsToTxnRequest;
import org.apache.kafka.common.requests.AddPartitionsToTxnResponse;
import org.apache.kafka.common.requests.EndTxnRequest;
import org.apache.kafka.common.requests.EndTxnResponse;
import org.apache.kafka.common.requests.FindCoordinatorRequest;
import org.apache.kafka.common.requests.FindCoordinatorResponse;
import org.apache.kafka.common.requests.InitProducerIdRequest;
import org.apache.kafka.common.requests.InitProducerIdResponse;
import org.apache.kafka.common.requests.ProduceResponse;
import org.apache.kafka.common.requests.TransactionResult;
import org.apache.kafka.common.requests.TxnOffsetCommitRequest;
import org.apache.kafka.common.requests.TxnOffsetCommitRequest.CommittedOffset;
import org.apache.kafka.common.requests.TxnOffsetCommitResponse;
import org.apache.kafka.common.utils.LogContext;
import org.slf4j.Logger;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.PriorityQueue;
import java.util.Set;

import static org.apache.kafka.common.record.RecordBatch.NO_PRODUCER_EPOCH;
import static org.apache.kafka.common.record.RecordBatch.NO_PRODUCER_ID;

/**
 * A class which maintains state for transactions. Also keeps the state necessary to ensure idempotent production.
 */
public class TransactionManager {
    private static final int NO_INFLIGHT_REQUEST_CORRELATION_ID = -1;

    private final Logger log;
    private final String transactionalId;
    private final int transactionTimeoutMs;

    // The base sequence of the next batch bound for a given partition.
    private final Map<TopicPartition, Integer> nextSequence;

    // The sequence of the last record of the last ack'd batch from the given partition. When there are no
    // in flight requests for a partition, the lastAckedSequence(topicPartition) == nextSequence(topicPartition) - 1.
    private final Map<TopicPartition, Integer> lastAckedSequence;

    // If a batch bound for a partition expired locally after being sent at least once, the partition has is considered
    // to have an unresolved state. We keep track fo such partitions here, and cannot assign any more sequence numbers
    // for this partition until the unresolved state gets cleared. This may happen if other inflight batches returned
    // successfully (indicating that the expired batch actually made it to the broker). If we don't get any successful
    // responses for the partition once the inflight request count falls to zero, we reset the producer id and
    // consequently clear this data structure as well.
    private final Set<TopicPartition> partitionsWithUnresolvedSequences;

    // Keep track of the in flight batches bound for a partition, ordered by sequence. This helps us to ensure that
    // we continue to order batches by the sequence numbers even when the responses come back out of order during
    // leader failover. We add a batch to the queue when it is drained, and remove it when the batch completes
    // (either successfully or through a fatal failure).
    private final Map<TopicPartition, PriorityQueue<ProducerBatch>> inflightBatchesBySequence;

    // We keep track of the last acknowledged offset on a per partition basis in order to disambiguate UnknownProducer
    // responses which are due to the retention period elapsing, and those which are due to actual lost data.
    private final Map<TopicPartition, Long> lastAckedOffset;

    private final PriorityQueue<TxnRequestHandler> pendingRequests;
    private final Set<TopicPartition> newPartitionsInTransaction;
    private final Set<TopicPartition> pendingPartitionsInTransaction;
    private final Set<TopicPartition> partitionsInTransaction;
    private final Map<TopicPartition, CommittedOffset> pendingTxnOffsetCommits;

    // This is used by the TxnRequestHandlers to control how long to back off before a given request is retried.
    // For instance, this value is lowered by the AddPartitionsToTxnHandler when it receives a CONCURRENT_TRANSACTIONS
    // error for the first AddPartitionsRequest in a transaction.
    private final long retryBackoffMs;

    // The retryBackoff is overridden to the following value if the first AddPartitions receives a
    // CONCURRENT_TRANSACTIONS error.
    private static final long ADD_PARTITIONS_RETRY_BACKOFF_MS = 20L;

    private int inFlightRequestCorrelationId = NO_INFLIGHT_REQUEST_CORRELATION_ID;
    private Node transactionCoordinator;
    private Node consumerGroupCoordinator;

    private volatile State currentState = State.UNINITIALIZED;
    private volatile RuntimeException lastError = null;
    private volatile ProducerIdAndEpoch producerIdAndEpoch;
    private volatile boolean transactionStarted = false;

    private enum State {
        UNINITIALIZED,
        INITIALIZING,
        READY,
        IN_TRANSACTION,
        COMMITTING_TRANSACTION,
        ABORTING_TRANSACTION,
        ABORTABLE_ERROR,
        FATAL_ERROR;

        private boolean isTransitionValid(State source, State target) {
            switch (target) {
                case INITIALIZING:
                    return source == UNINITIALIZED;
                case READY:
                    return source == INITIALIZING || source == COMMITTING_TRANSACTION || source == ABORTING_TRANSACTION;
                case IN_TRANSACTION:
                    return source == READY;
                case COMMITTING_TRANSACTION:
                    return source == IN_TRANSACTION;
                case ABORTING_TRANSACTION:
                    return source == IN_TRANSACTION || source == ABORTABLE_ERROR;
                case ABORTABLE_ERROR:
                    return source == IN_TRANSACTION || source == COMMITTING_TRANSACTION || source == ABORTABLE_ERROR;
                case FATAL_ERROR:
                default:
                    // We can transition to FATAL_ERROR unconditionally.
                    // FATAL_ERROR is never a valid starting state for any transition. So the only option is to close the
                    // producer or do purely non transactional requests.
                    return true;
            }
        }
    }

    // We use the priority to determine the order in which requests need to be sent out. For instance, if we have
    // a pending FindCoordinator request, that must always go first. Next, If we need a producer id, that must go second.
    // The endTxn request must always go last.
    private enum Priority {
        FIND_COORDINATOR(0),
        INIT_PRODUCER_ID(1),
        ADD_PARTITIONS_OR_OFFSETS(2),
        END_TXN(3);

        final int priority;

        Priority(int priority) {
            this.priority = priority;
        }
    }

    public TransactionManager(LogContext logContext, String transactionalId, int transactionTimeoutMs, long retryBackoffMs) {
        this.producerIdAndEpoch = new ProducerIdAndEpoch(NO_PRODUCER_ID, NO_PRODUCER_EPOCH);
        this.nextSequence = new HashMap<>();
        this.lastAckedSequence = new HashMap<>();
        this.transactionalId = transactionalId;
        this.log = logContext.logger(TransactionManager.class);
        this.transactionTimeoutMs = transactionTimeoutMs;
        this.transactionCoordinator = null;
        this.consumerGroupCoordinator = null;
        this.newPartitionsInTransaction = new HashSet<>();
        this.pendingPartitionsInTransaction = new HashSet<>();
        this.partitionsInTransaction = new HashSet<>();
        this.pendingTxnOffsetCommits = new HashMap<>();
        this.pendingRequests = new PriorityQueue<>(10, new Comparator<TxnRequestHandler>() {
            @Override
            public int compare(TxnRequestHandler o1, TxnRequestHandler o2) {
                return Integer.compare(o1.priority().priority, o2.priority().priority);
            }
        });

        this.partitionsWithUnresolvedSequences = new HashSet<>();
        this.inflightBatchesBySequence = new HashMap<>();
        this.lastAckedOffset = new HashMap<>();

        this.retryBackoffMs = retryBackoffMs;
    }

    TransactionManager() {
        this(new LogContext(), null, 0, 100);
    }

    public synchronized TransactionalRequestResult initializeTransactions() {
        ensureTransactional();
        transitionTo(State.INITIALIZING);
        setProducerIdAndEpoch(ProducerIdAndEpoch.NONE);
        this.nextSequence.clear();
        InitProducerIdRequest.Builder builder = new InitProducerIdRequest.Builder(transactionalId, transactionTimeoutMs);
        InitProducerIdHandler handler = new InitProducerIdHandler(builder);
        enqueueRequest(handler);
        return handler.result;
    }

    public synchronized void beginTransaction() {
        ensureTransactional();
        maybeFailWithError();
        transitionTo(State.IN_TRANSACTION);
    }

    public synchronized TransactionalRequestResult beginCommit() {
        ensureTransactional();
        maybeFailWithError();
        transitionTo(State.COMMITTING_TRANSACTION);
        return beginCompletingTransaction(TransactionResult.COMMIT);
    }

    public synchronized TransactionalRequestResult beginAbort() {
        ensureTransactional();
        if (currentState != State.ABORTABLE_ERROR)
            maybeFailWithError();
        transitionTo(State.ABORTING_TRANSACTION);

        // We're aborting the transaction, so there should be no need to add new partitions
        newPartitionsInTransaction.clear();
        return beginCompletingTransaction(TransactionResult.ABORT);
    }

    private TransactionalRequestResult beginCompletingTransaction(TransactionResult transactionResult) {
        if (!newPartitionsInTransaction.isEmpty())
            enqueueRequest(addPartitionsToTransactionHandler());
        EndTxnRequest.Builder builder = new EndTxnRequest.Builder(transactionalId, producerIdAndEpoch.producerId,
                producerIdAndEpoch.epoch, transactionResult);
        EndTxnHandler handler = new EndTxnHandler(builder);
        enqueueRequest(handler);
        return handler.result;
    }

    public synchronized TransactionalRequestResult sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
                                                                            String consumerGroupId) {
        ensureTransactional();
        maybeFailWithError();
        if (currentState != State.IN_TRANSACTION)
            throw new KafkaException("Cannot send offsets to transaction either because the producer is not in an " +
                    "active transaction");

        log.debug("Begin adding offsets {} for consumer group {} to transaction", offsets, consumerGroupId);
        AddOffsetsToTxnRequest.Builder builder = new AddOffsetsToTxnRequest.Builder(transactionalId,
                producerIdAndEpoch.producerId, producerIdAndEpoch.epoch, consumerGroupId);
        AddOffsetsToTxnHandler handler = new AddOffsetsToTxnHandler(builder, offsets);
        enqueueRequest(handler);
        return handler.result;
    }

    public synchronized void maybeAddPartitionToTransaction(TopicPartition topicPartition) {
        failIfNotReadyForSend();

        if (isPartitionAdded(topicPartition) || isPartitionPendingAdd(topicPartition))
            return;

        log.debug("Begin adding new partition {} to transaction", topicPartition);
        newPartitionsInTransaction.add(topicPartition);
    }

    RuntimeException lastError() {
        return lastError;
    }

    public synchronized void failIfNotReadyForSend() {
        if (hasError())
            throw new KafkaException("Cannot perform send because at least one previous transactional or " +
                    "idempotent request has failed with errors.", lastError);

        if (isTransactional()) {
            if (!hasProducerId())
                throw new IllegalStateException("Cannot perform a 'send' before completing a call to initTransactions " +
                        "when transactions are enabled.");

            if (currentState != State.IN_TRANSACTION)
                throw new IllegalStateException("Cannot call send in state " + currentState);
        }
    }

    synchronized boolean isSendToPartitionAllowed(TopicPartition tp) {
        if (hasFatalError())
            return false;
        return !isTransactional() || partitionsInTransaction.contains(tp);
    }

    public String transactionalId() {
        return transactionalId;
    }

    public boolean hasProducerId() {
        return producerIdAndEpoch.isValid();
    }

    public boolean isTransactional() {
        return transactionalId != null;
    }

    synchronized boolean hasPartitionsToAdd() {
        return !newPartitionsInTransaction.isEmpty() || !pendingPartitionsInTransaction.isEmpty();
    }

    synchronized boolean isCompleting() {
        return currentState == State.COMMITTING_TRANSACTION || currentState == State.ABORTING_TRANSACTION;
    }

    synchronized boolean hasError() {
        return currentState == State.ABORTABLE_ERROR || currentState == State.FATAL_ERROR;
    }

    synchronized boolean isAborting() {
        return currentState == State.ABORTING_TRANSACTION;
    }

    synchronized void transitionToAbortableError(RuntimeException exception) {
        if (currentState == State.ABORTING_TRANSACTION) {
            log.debug("Skipping transition to abortable error state since the transaction is already being " +
                    "aborted. Underlying exception: ", exception);
            return;
        }
        transitionTo(State.ABORTABLE_ERROR, exception);
    }

    synchronized void transitionToFatalError(RuntimeException exception) {
        transitionTo(State.FATAL_ERROR, exception);
    }

    // visible for testing
    synchronized boolean isPartitionAdded(TopicPartition partition) {
        return partitionsInTransaction.contains(partition);
    }

    // visible for testing
    synchronized boolean isPartitionPendingAdd(TopicPartition partition) {
        return newPartitionsInTransaction.contains(partition) || pendingPartitionsInTransaction.contains(partition);
    }

    /**
     * Get the current producer id and epoch without blocking. Callers must use {@link ProducerIdAndEpoch#isValid()} to
     * verify that the result is valid.
     *
     * @return the current ProducerIdAndEpoch.
     */
    ProducerIdAndEpoch producerIdAndEpoch() {
        return producerIdAndEpoch;
    }

    boolean hasProducerId(long producerId) {
        return producerIdAndEpoch.producerId == producerId;
    }

    boolean hasProducerIdAndEpoch(long producerId, short producerEpoch) {
        ProducerIdAndEpoch idAndEpoch = this.producerIdAndEpoch;
        return idAndEpoch.producerId == producerId && idAndEpoch.epoch == producerEpoch;
    }

    /**
     * Set the producer id and epoch atomically.
     */
    void setProducerIdAndEpoch(ProducerIdAndEpoch producerIdAndEpoch) {
        log.info("ProducerId set to {} with epoch {}", producerIdAndEpoch.producerId, producerIdAndEpoch.epoch);
        this.producerIdAndEpoch = producerIdAndEpoch;
    }

    /**
     * This method is used when the producer needs to reset its internal state because of an irrecoverable exception
     * from the broker.
     *
     * We need to reset the producer id and associated state when we have sent a batch to the broker, but we either get
     * a non-retriable exception or we run out of retries, or the batch expired in the producer queue after it was already
     * sent to the broker.
     *
     * In all of these cases, we don't know whether batch was actually committed on the broker, and hence whether the
     * sequence number was actually updated. If we don't reset the producer state, we risk the chance that all future
     * messages will return an OutOfOrderSequenceException.
     *
     * Note that we can't reset the producer state for the transactional producer as this would mean bumping the epoch
     * for the same producer id. This might involve aborting the ongoing transaction during the initPidRequest, and the user
     * would not have any way of knowing this happened. So for the transactional producer, it's best to return the
     * produce error to the user and let them abort the transaction and close the producer explicitly.
     */
    synchronized void resetProducerId() {
        if (isTransactional())
            throw new IllegalStateException("Cannot reset producer state for a transactional producer. " +
                    "You must either abort the ongoing transaction or reinitialize the transactional producer instead");
        setProducerIdAndEpoch(ProducerIdAndEpoch.NONE);
        this.nextSequence.clear();
        this.lastAckedSequence.clear();
        this.inflightBatchesBySequence.clear();
        this.partitionsWithUnresolvedSequences.clear();
        this.lastAckedOffset.clear();
    }

    /**
     * Returns the next sequence number to be written to the given TopicPartition.
     */
    synchronized Integer sequenceNumber(TopicPartition topicPartition) {
        Integer currentSequenceNumber = nextSequence.get(topicPartition);
        if (currentSequenceNumber == null) {
            currentSequenceNumber = 0;
            nextSequence.put(topicPartition, currentSequenceNumber);
        }
        return currentSequenceNumber;
    }

    synchronized void incrementSequenceNumber(TopicPartition topicPartition, int increment) {
        Integer currentSequenceNumber = nextSequence.get(topicPartition);
        if (currentSequenceNumber == null)
            throw new IllegalStateException("Attempt to increment sequence number for a partition with no current sequence.");

        currentSequenceNumber += increment;
        nextSequence.put(topicPartition, currentSequenceNumber);
    }

    synchronized void addInFlightBatch(ProducerBatch batch) {
        if (!batch.hasSequence())
            throw new IllegalStateException("Can't track batch for partition " + batch.topicPartition + " when sequence is not set.");
        if (!inflightBatchesBySequence.containsKey(batch.topicPartition)) {
            inflightBatchesBySequence.put(batch.topicPartition, new PriorityQueue<>(5, new Comparator<ProducerBatch>() {
                @Override
                public int compare(ProducerBatch o1, ProducerBatch o2) {
                    return o1.baseSequence() - o2.baseSequence();
                }
            }));
        }
        inflightBatchesBySequence.get(batch.topicPartition).offer(batch);
    }

    /**
     * Returns the first inflight sequence for a given partition. This is the base sequence of an inflight batch with
     * the lowest sequence number.
     * @return the lowest inflight sequence if the transaction manager is tracking inflight requests for this partition.
     *         If there are no inflight requests being tracked for this partition, this method will return
     *         RecordBatch.NO_SEQUENCE.
     */
    synchronized int firstInFlightSequence(TopicPartition topicPartition) {
        PriorityQueue<ProducerBatch> inFlightBatches = inflightBatchesBySequence.get(topicPartition);
        if (inFlightBatches == null)
            return RecordBatch.NO_SEQUENCE;

        ProducerBatch firstInFlightBatch = inFlightBatches.peek();
        if (firstInFlightBatch == null)
            return RecordBatch.NO_SEQUENCE;

        return firstInFlightBatch.baseSequence();
    }

    synchronized ProducerBatch nextBatchBySequence(TopicPartition topicPartition) {
        PriorityQueue<ProducerBatch> queue = inflightBatchesBySequence.get(topicPartition);
        if (queue == null)
            return null;
        return queue.peek();
    }

    synchronized void removeInFlightBatch(ProducerBatch batch) {
        PriorityQueue<ProducerBatch> queue = inflightBatchesBySequence.get(batch.topicPartition);
        if (queue == null)
            return;
        queue.remove(batch);
    }

    synchronized void maybeUpdateLastAckedSequence(TopicPartition topicPartition, int sequence) {
        if (sequence > lastAckedSequence(topicPartition))
            lastAckedSequence.put(topicPartition, sequence);
    }

    synchronized int lastAckedSequence(TopicPartition topicPartition) {
        Integer currentLastAckedSequence = lastAckedSequence.get(topicPartition);
        if (currentLastAckedSequence == null)
            return -1;
        return currentLastAckedSequence;
    }

    synchronized long lastAckedOffset(TopicPartition topicPartition) {
        Long offset = lastAckedOffset.get(topicPartition);
        if (offset == null)
            return ProduceResponse.INVALID_OFFSET;
        return offset;
    }

    synchronized void updateLastAckedOffset(ProduceResponse.PartitionResponse response, ProducerBatch batch) {
        if (response.baseOffset == ProduceResponse.INVALID_OFFSET)
            return;
        long lastOffset = response.baseOffset + batch.recordCount - 1;
        if (lastOffset > lastAckedOffset(batch.topicPartition)) {
            lastAckedOffset.put(batch.topicPartition, lastOffset);
        } else {
            log.trace("Partition {} keeps lastOffset at {}", batch.topicPartition, lastOffset);
        }
    }

    // If a batch is failed fatally, the sequence numbers for future batches bound for the partition must be adjusted
    // so that they don't fail with the OutOfOrderSequenceException.
    //
    // This method must only be called when we know that the batch is question has been unequivocally failed by the broker,
    // ie. it has received a confirmed fatal status code like 'Message Too Large' or something similar.
    synchronized void adjustSequencesDueToFailedBatch(ProducerBatch batch) {
        if (!this.nextSequence.containsKey(batch.topicPartition))
            // Sequence numbers are not being tracked for this partition. This could happen if the producer id was just
            // reset due to a previous OutOfOrderSequenceException.
            return;
        log.debug("producerId: {}, send to partition {} failed fatally. Reducing future sequence numbers by {}",
                batch.producerId(), batch.topicPartition, batch.recordCount);
        int currentSequence = sequenceNumber(batch.topicPartition);
        currentSequence -= batch.recordCount;
        if (currentSequence < 0)
            throw new IllegalStateException("Sequence number for partition " + batch.topicPartition + " is going to become negative : " + currentSequence);

        setNextSequence(batch.topicPartition, currentSequence);

        for (ProducerBatch inFlightBatch : inflightBatchesBySequence.get(batch.topicPartition)) {
            if (inFlightBatch.baseSequence() < batch.baseSequence())
                continue;
            int newSequence = inFlightBatch.baseSequence() - batch.recordCount;
            if (newSequence < 0)
                throw new IllegalStateException("Sequence number for batch with sequence " + inFlightBatch.baseSequence()
                        + " for partition " + batch.topicPartition + " is going to become negative :" + newSequence);

            log.info("Resetting sequence number of batch with current sequence {} for partition {} to {}", inFlightBatch.baseSequence(), batch.topicPartition, newSequence);
            inFlightBatch.resetProducerState(new ProducerIdAndEpoch(inFlightBatch.producerId(), inFlightBatch.producerEpoch()), newSequence, inFlightBatch.isTransactional());
        }
    }

    private synchronized void startSequencesAtBeginning(TopicPartition topicPartition) {
        int sequence = 0;
        for (ProducerBatch inFlightBatch : inflightBatchesBySequence.get(topicPartition)) {
            log.info("Resetting sequence number of batch with current sequence {} for partition {} to {}",
                    inFlightBatch.baseSequence(), inFlightBatch.topicPartition, sequence);
            inFlightBatch.resetProducerState(new ProducerIdAndEpoch(inFlightBatch.producerId(),
                    inFlightBatch.producerEpoch()), sequence, inFlightBatch.isTransactional());

            sequence += inFlightBatch.recordCount;
        }
        setNextSequence(topicPartition, sequence);
        lastAckedSequence.remove(topicPartition);
    }

    synchronized boolean hasInflightBatches(TopicPartition topicPartition) {
        return inflightBatchesBySequence.containsKey(topicPartition) && !inflightBatchesBySequence.get(topicPartition).isEmpty();
    }

    synchronized boolean hasUnresolvedSequences() {
        return !partitionsWithUnresolvedSequences.isEmpty();
    }

    synchronized boolean hasUnresolvedSequence(TopicPartition topicPartition) {
        return partitionsWithUnresolvedSequences.contains(topicPartition);
    }

    synchronized void markSequenceUnresolved(TopicPartition topicPartition) {
        log.debug("Marking partition {} unresolved", topicPartition);
        partitionsWithUnresolvedSequences.add(topicPartition);
    }

    // Checks if there are any partitions with unresolved partitions which may now be resolved. Returns true if
    // the producer id needs a reset, false otherwise.
    synchronized boolean shouldResetProducerStateAfterResolvingSequences() {
        if (isTransactional())
            // We should not reset producer state if we are transactional. We will transition to a fatal error instead.
            return false;
        for (Iterator<TopicPartition> iter = partitionsWithUnresolvedSequences.iterator(); iter.hasNext(); ) {
            TopicPartition topicPartition = iter.next();
            if (!hasInflightBatches(topicPartition)) {
                // The partition has been fully drained. At this point, the last ack'd sequence should be once less than
                // next sequence destined for the partition. If so, the partition is fully resolved. If not, we should
                // reset the sequence number if necessary.
                if (isNextSequence(topicPartition, sequenceNumber(topicPartition))) {
                    // This would happen when a batch was expired, but subsequent batches succeeded.
                    iter.remove();
                } else {
                    // We would enter this branch if all in flight batches were ultimately expired in the producer.
                    log.info("No inflight batches remaining for {}, last ack'd sequence for partition is {}, next sequence is {}. " +
                            "Going to reset producer state.", topicPartition, lastAckedSequence(topicPartition), sequenceNumber(topicPartition));
                    return true;
                }
            }
        }
        return false;
    }

    synchronized boolean isNextSequence(TopicPartition topicPartition, int sequence) {
        return sequence - lastAckedSequence(topicPartition) == 1;
    }

    private synchronized void setNextSequence(TopicPartition topicPartition, int sequence) {
        if (!nextSequence.containsKey(topicPartition) && sequence != 0)
            throw new IllegalStateException("Trying to set the sequence number for " + topicPartition + " to " + sequence +
            ", but the sequence number was never set for this partition.");
        nextSequence.put(topicPartition, sequence);
    }

    synchronized TxnRequestHandler nextRequestHandler(boolean hasIncompleteBatches) {
        if (!newPartitionsInTransaction.isEmpty())
            enqueueRequest(addPartitionsToTransactionHandler());

        TxnRequestHandler nextRequestHandler = pendingRequests.peek();
        if (nextRequestHandler == null)
            return null;

        // Do not send the EndTxn until all batches have been flushed
        if (nextRequestHandler.isEndTxn() && hasIncompleteBatches)
            return null;

        pendingRequests.poll();
        if (maybeTerminateRequestWithError(nextRequestHandler)) {
            log.trace("Not sending transactional request {} because we are in an error state",
                    nextRequestHandler.requestBuilder());
            return null;
        }

        if (nextRequestHandler.isEndTxn() && !transactionStarted) {
            nextRequestHandler.result.done();
            if (currentState != State.FATAL_ERROR) {
                log.debug("Not sending EndTxn for completed transaction since no partitions " +
                        "or offsets were successfully added");
                completeTransaction();
            }
            nextRequestHandler = pendingRequests.poll();
        }

        if (nextRequestHandler != null)
            log.trace("Request {} dequeued for sending", nextRequestHandler.requestBuilder());

        return nextRequestHandler;
    }

    synchronized void retry(TxnRequestHandler request) {
        request.setRetry();
        enqueueRequest(request);
    }

    synchronized void authenticationFailed(AuthenticationException e) {
        for (TxnRequestHandler request : pendingRequests)
            request.fatalError(e);
    }

    Node coordinator(FindCoordinatorRequest.CoordinatorType type) {
        switch (type) {
            case GROUP:
                return consumerGroupCoordinator;
            case TRANSACTION:
                return transactionCoordinator;
            default:
                throw new IllegalStateException("Received an invalid coordinator type: " + type);
        }
    }

    void lookupCoordinator(TxnRequestHandler request) {
        lookupCoordinator(request.coordinatorType(), request.coordinatorKey());
    }

    void setInFlightTransactionalRequestCorrelationId(int correlationId) {
        inFlightRequestCorrelationId = correlationId;
    }

    void clearInFlightTransactionalRequestCorrelationId() {
        inFlightRequestCorrelationId = NO_INFLIGHT_REQUEST_CORRELATION_ID;
    }

    boolean hasInFlightTransactionalRequest() {
        return inFlightRequestCorrelationId != NO_INFLIGHT_REQUEST_CORRELATION_ID;
    }

    // visible for testing.
    boolean hasFatalError() {
        return currentState == State.FATAL_ERROR;
    }

    // visible for testing.
    boolean hasAbortableError() {
        return currentState == State.ABORTABLE_ERROR;
    }

    // visible for testing
    synchronized boolean transactionContainsPartition(TopicPartition topicPartition) {
        return partitionsInTransaction.contains(topicPartition);
    }

    // visible for testing
    synchronized boolean hasPendingOffsetCommits() {
        return !pendingTxnOffsetCommits.isEmpty();
    }

    // visible for testing
    synchronized boolean hasOngoingTransaction() {
        // transactions are considered ongoing once started until completion or a fatal error
        return currentState == State.IN_TRANSACTION || isCompleting() || hasAbortableError();
    }

    synchronized boolean canRetry(ProduceResponse.PartitionResponse response, ProducerBatch batch) {
        if (!hasProducerId(batch.producerId()))
            return false;

        Errors error = response.error;
        if (error == Errors.OUT_OF_ORDER_SEQUENCE_NUMBER && !hasUnresolvedSequence(batch.topicPartition) &&
                (batch.sequenceHasBeenReset() || !isNextSequence(batch.topicPartition, batch.baseSequence())))
            // We should retry the OutOfOrderSequenceException if the batch is _not_ the next batch, ie. its base
            // sequence isn't the lastAckedSequence + 1. However, if the first in flight batch fails fatally, we will
            // adjust the sequences of the other inflight batches to account for the 'loss' of the sequence range in
            // the batch which failed. In this case, an inflight batch will have a base sequence which is
            // the lastAckedSequence + 1 after adjustment. When this batch fails with an OutOfOrderSequence, we want to retry it.
            // To account for the latter case, we check whether the sequence has been reset since the last drain.
            // If it has, we will retry it anyway.
            return true;

        if (error == Errors.UNKNOWN_PRODUCER_ID) {
            if (response.logStartOffset == -1)
                // We don't know the log start offset with this response. We should just retry the request until we get it.
                // The UNKNOWN_PRODUCER_ID error code was added along with the new ProduceResponse which includes the
                // logStartOffset. So the '-1' sentinel is not for backward compatibility. Instead, it is possible for
                // a broker to not know the logStartOffset at when it is returning the response because the partition
                // may have moved away from the broker from the time the error was initially raised to the time the
                // response was being constructed. In these cases, we should just retry the request: we are guaranteed
                // to eventually get a logStartOffset once things settle down.
                return true;

            if (batch.sequenceHasBeenReset()) {
                // When the first inflight batch fails due to the truncation case, then the sequences of all the other
                // in flight batches would have been restarted from the beginning. However, when those responses
                // come back from the broker, they would also come with an UNKNOWN_PRODUCER_ID error. In this case, we should not
                // reset the sequence numbers to the beginning.
                return true;
            } else if (lastAckedOffset(batch.topicPartition) < response.logStartOffset) {
                // The head of the log has been removed, probably due to the retention time elapsing. In this case,
                // we expect to lose the producer state. Reset the sequences of all inflight batches to be from the beginning
                // and retry them.
                startSequencesAtBeginning(batch.topicPartition);
                return true;
            }
        }
        return false;
    }

    // visible for testing
    synchronized boolean isReady() {
        return isTransactional() && currentState == State.READY;
    }

    private void transitionTo(State target) {
        transitionTo(target, null);
    }

    private synchronized void transitionTo(State target, RuntimeException error) {
        if (!currentState.isTransitionValid(currentState, target)) {
            String idString = transactionalId == null ?  "" : "TransactionalId " + transactionalId + ": ";
            throw new KafkaException(idString + "Invalid transition attempted from state "
                    + currentState.name() + " to state " + target.name());
        }

        if (target == State.FATAL_ERROR || target == State.ABORTABLE_ERROR) {
            if (error == null)
                throw new IllegalArgumentException("Cannot transition to " + target + " with an null exception");
            lastError = error;
        } else {
            lastError = null;
        }

        if (lastError != null)
            log.debug("Transition from state {} to error state {}", currentState, target, lastError);
        else
            log.debug("Transition from state {} to {}", currentState, target);

        currentState = target;
    }

    private void ensureTransactional() {
        if (!isTransactional())
            throw new IllegalStateException("Transactional method invoked on a non-transactional producer.");
    }

    private void maybeFailWithError() {
        if (hasError())
            throw new KafkaException("Cannot execute transactional method because we are in an error state", lastError);
    }

    private boolean maybeTerminateRequestWithError(TxnRequestHandler requestHandler) {
        if (hasError()) {
            if (hasAbortableError() && requestHandler instanceof FindCoordinatorHandler)
                // No harm letting the FindCoordinator request go through if we're expecting to abort
                return false;

            requestHandler.fail(lastError);
            return true;
        }
        return false;
    }

    private void enqueueRequest(TxnRequestHandler requestHandler) {
        log.debug("Enqueuing transactional request {}", requestHandler.requestBuilder());
        pendingRequests.add(requestHandler);
    }

    private synchronized void lookupCoordinator(FindCoordinatorRequest.CoordinatorType type, String coordinatorKey) {
        switch (type) {
            case GROUP:
                consumerGroupCoordinator = null;
                break;
            case TRANSACTION:
                transactionCoordinator = null;
                break;
            default:
                throw new IllegalStateException("Invalid coordinator type: " + type);
        }

        FindCoordinatorRequest.Builder builder = new FindCoordinatorRequest.Builder(type, coordinatorKey);
        enqueueRequest(new FindCoordinatorHandler(builder));
    }

    private synchronized void completeTransaction() {
        transitionTo(State.READY);
        lastError = null;
        transactionStarted = false;
        newPartitionsInTransaction.clear();
        pendingPartitionsInTransaction.clear();
        partitionsInTransaction.clear();
    }

    private synchronized TxnRequestHandler addPartitionsToTransactionHandler() {
        pendingPartitionsInTransaction.addAll(newPartitionsInTransaction);
        newPartitionsInTransaction.clear();
        AddPartitionsToTxnRequest.Builder builder = new AddPartitionsToTxnRequest.Builder(transactionalId,
                producerIdAndEpoch.producerId, producerIdAndEpoch.epoch, new ArrayList<>(pendingPartitionsInTransaction));
        return new AddPartitionsToTxnHandler(builder);
    }

    private TxnOffsetCommitHandler txnOffsetCommitHandler(TransactionalRequestResult result,
                                                          Map<TopicPartition, OffsetAndMetadata> offsets,
                                                          String consumerGroupId) {
        for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsets.entrySet()) {
            OffsetAndMetadata offsetAndMetadata = entry.getValue();
            CommittedOffset committedOffset = new CommittedOffset(offsetAndMetadata.offset(), offsetAndMetadata.metadata());
            pendingTxnOffsetCommits.put(entry.getKey(), committedOffset);
        }
        TxnOffsetCommitRequest.Builder builder = new TxnOffsetCommitRequest.Builder(transactionalId, consumerGroupId,
                producerIdAndEpoch.producerId, producerIdAndEpoch.epoch, pendingTxnOffsetCommits);
        return new TxnOffsetCommitHandler(result, builder);
    }

    abstract class TxnRequestHandler implements RequestCompletionHandler {
        protected final TransactionalRequestResult result;
        private boolean isRetry = false;

        TxnRequestHandler(TransactionalRequestResult result) {
            this.result = result;
        }

        TxnRequestHandler() {
            this(new TransactionalRequestResult());
        }

        void fatalError(RuntimeException e) {
            result.setError(e);
            transitionToFatalError(e);
            result.done();
        }

        void abortableError(RuntimeException e) {
            result.setError(e);
            transitionToAbortableError(e);
            result.done();
        }

        void fail(RuntimeException e) {
            result.setError(e);
            result.done();
        }

        void reenqueue() {
            synchronized (TransactionManager.this) {
                this.isRetry = true;
                enqueueRequest(this);
            }
        }

        long retryBackoffMs() {
            return retryBackoffMs;
        }

        @Override
        public void onComplete(ClientResponse response) {
            if (response.requestHeader().correlationId() != inFlightRequestCorrelationId) {
                fatalError(new RuntimeException("Detected more than one in-flight transactional request."));
            } else {
                clearInFlightTransactionalRequestCorrelationId();
                if (response.wasDisconnected()) {
                    log.debug("Disconnected from {}. Will retry.", response.destination());
                    if (this.needsCoordinator())
                        lookupCoordinator(this.coordinatorType(), this.coordinatorKey());
                    reenqueue();
                } else if (response.versionMismatch() != null) {
                    fatalError(response.versionMismatch());
                } else if (response.hasResponse()) {
                    log.trace("Received transactional response {} for request {}", response.responseBody(),
                            requestBuilder());
                    synchronized (TransactionManager.this) {
                        handleResponse(response.responseBody());
                    }
                } else {
                    fatalError(new KafkaException("Could not execute transactional request for unknown reasons"));
                }
            }
        }

        boolean needsCoordinator() {
            return coordinatorType() != null;
        }

        FindCoordinatorRequest.CoordinatorType coordinatorType() {
            return FindCoordinatorRequest.CoordinatorType.TRANSACTION;
        }

        String coordinatorKey() {
            return transactionalId;
        }

        void setRetry() {
            this.isRetry = true;
        }

        boolean isRetry() {
            return isRetry;
        }

        boolean isEndTxn() {
            return false;
        }

        abstract AbstractRequest.Builder<?> requestBuilder();

        abstract void handleResponse(AbstractResponse responseBody);

        abstract Priority priority();
    }

    private class InitProducerIdHandler extends TxnRequestHandler {
        private final InitProducerIdRequest.Builder builder;

        private InitProducerIdHandler(InitProducerIdRequest.Builder builder) {
            this.builder = builder;
        }

        @Override
        InitProducerIdRequest.Builder requestBuilder() {
            return builder;
        }

        @Override
        Priority priority() {
            return Priority.INIT_PRODUCER_ID;
        }

        @Override
        public void handleResponse(AbstractResponse response) {
            InitProducerIdResponse initProducerIdResponse = (InitProducerIdResponse) response;
            Errors error = initProducerIdResponse.error();

            if (error == Errors.NONE) {
                ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(initProducerIdResponse.producerId(), initProducerIdResponse.epoch());
                setProducerIdAndEpoch(producerIdAndEpoch);
                transitionTo(State.READY);
                lastError = null;
                result.done();
            } else if (error == Errors.NOT_COORDINATOR || error == Errors.COORDINATOR_NOT_AVAILABLE) {
                lookupCoordinator(FindCoordinatorRequest.CoordinatorType.TRANSACTION, transactionalId);
                reenqueue();
            } else if (error == Errors.COORDINATOR_LOAD_IN_PROGRESS || error == Errors.CONCURRENT_TRANSACTIONS) {
                reenqueue();
            } else if (error == Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED) {
                fatalError(error.exception());
            } else {
                fatalError(new KafkaException("Unexpected error in InitProducerIdResponse; " + error.message()));
            }
        }
    }

    private class AddPartitionsToTxnHandler extends TxnRequestHandler {
        private final AddPartitionsToTxnRequest.Builder builder;
        private long retryBackoffMs;

        private AddPartitionsToTxnHandler(AddPartitionsToTxnRequest.Builder builder) {
            this.builder = builder;
            this.retryBackoffMs = TransactionManager.this.retryBackoffMs;
        }

        @Override
        AddPartitionsToTxnRequest.Builder requestBuilder() {
            return builder;
        }

        @Override
        Priority priority() {
            return Priority.ADD_PARTITIONS_OR_OFFSETS;
        }

        @Override
        public void handleResponse(AbstractResponse response) {
            AddPartitionsToTxnResponse addPartitionsToTxnResponse = (AddPartitionsToTxnResponse) response;
            Map<TopicPartition, Errors> errors = addPartitionsToTxnResponse.errors();
            boolean hasPartitionErrors = false;
            Set<String> unauthorizedTopics = new HashSet<>();
            retryBackoffMs = TransactionManager.this.retryBackoffMs;

            for (Map.Entry<TopicPartition, Errors> topicPartitionErrorEntry : errors.entrySet()) {
                TopicPartition topicPartition = topicPartitionErrorEntry.getKey();
                Errors error = topicPartitionErrorEntry.getValue();

                if (error == Errors.NONE) {
                    continue;
                } else if (error == Errors.COORDINATOR_NOT_AVAILABLE || error == Errors.NOT_COORDINATOR) {
                    lookupCoordinator(FindCoordinatorRequest.CoordinatorType.TRANSACTION, transactionalId);
                    reenqueue();
                    return;
                } else if (error == Errors.CONCURRENT_TRANSACTIONS) {
                    maybeOverrideRetryBackoffMs();
                    reenqueue();
                    return;
                } else if (error == Errors.COORDINATOR_LOAD_IN_PROGRESS || error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
                    reenqueue();
                    return;
                } else if (error == Errors.INVALID_PRODUCER_EPOCH) {
                    fatalError(error.exception());
                    return;
                } else if (error == Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED) {
                    fatalError(error.exception());
                    return;
                } else if (error == Errors.INVALID_PRODUCER_ID_MAPPING
                        || error == Errors.INVALID_TXN_STATE) {
                    fatalError(new KafkaException(error.exception()));
                    return;
                } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
                    unauthorizedTopics.add(topicPartition.topic());
                } else if (error == Errors.OPERATION_NOT_ATTEMPTED) {
                    log.debug("Did not attempt to add partition {} to transaction because other partitions in the " +
                            "batch had errors.", topicPartition);
                    hasPartitionErrors = true;
                } else {
                    log.error("Could not add partition {} due to unexpected error {}", topicPartition, error);
                    hasPartitionErrors = true;
                }
            }

            Set<TopicPartition> partitions = errors.keySet();

            // Remove the partitions from the pending set regardless of the result. We use the presence
            // of partitions in the pending set to know when it is not safe to send batches. However, if
            // the partitions failed to be added and we enter an error state, we expect the batches to be
            // aborted anyway. In this case, we must be able to continue sending the batches which are in
            // retry for partitions that were successfully added.
            pendingPartitionsInTransaction.removeAll(partitions);

            if (!unauthorizedTopics.isEmpty()) {
                abortableError(new TopicAuthorizationException(unauthorizedTopics));
            } else if (hasPartitionErrors) {
                abortableError(new KafkaException("Could not add partitions to transaction due to errors: " + errors));
            } else {
                log.debug("Successfully added partitions {} to transaction", partitions);
                partitionsInTransaction.addAll(partitions);
                transactionStarted = true;
                result.done();
            }
        }

        @Override
        public long retryBackoffMs() {
            return Math.min(TransactionManager.this.retryBackoffMs, this.retryBackoffMs);
        }

        private void maybeOverrideRetryBackoffMs() {
            // We only want to reduce the backoff when retrying the first AddPartition which errored out due to a
            // CONCURRENT_TRANSACTIONS error since this means that the previous transaction is still completing and
            // we don't want to wait too long before trying to start the new one.
            //
            // This is only a temporary fix, the long term solution is being tracked in
            // https://issues.apache.org/jira/browse/KAFKA-5482
            if (partitionsInTransaction.isEmpty())
                this.retryBackoffMs = ADD_PARTITIONS_RETRY_BACKOFF_MS;
        }
    }

    private class FindCoordinatorHandler extends TxnRequestHandler {
        private final FindCoordinatorRequest.Builder builder;

        private FindCoordinatorHandler(FindCoordinatorRequest.Builder builder) {
            this.builder = builder;
        }

        @Override
        FindCoordinatorRequest.Builder requestBuilder() {
            return builder;
        }

        @Override
        Priority priority() {
            return Priority.FIND_COORDINATOR;
        }

        @Override
        FindCoordinatorRequest.CoordinatorType coordinatorType() {
            return null;
        }

        @Override
        String coordinatorKey() {
            return null;
        }

        @Override
        public void handleResponse(AbstractResponse response) {
            FindCoordinatorResponse findCoordinatorResponse = (FindCoordinatorResponse) response;
            Errors error = findCoordinatorResponse.error();

            if (error == Errors.NONE) {
                Node node = findCoordinatorResponse.node();
                switch (builder.coordinatorType()) {
                    case GROUP:
                        consumerGroupCoordinator = node;
                        break;
                    case TRANSACTION:
                        transactionCoordinator = node;
                }
                result.done();
            } else if (error == Errors.COORDINATOR_NOT_AVAILABLE) {
                reenqueue();
            } else if (error == Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED) {
                fatalError(error.exception());
            } else if (findCoordinatorResponse.error() == Errors.GROUP_AUTHORIZATION_FAILED) {
                abortableError(new GroupAuthorizationException(builder.coordinatorKey()));
            } else {
                fatalError(new KafkaException(String.format("Could not find a coordinator with type %s with key %s due to" +
                        "unexpected error: %s", builder.coordinatorType(), builder.coordinatorKey(),
                        findCoordinatorResponse.error().message())));
            }
        }
    }

    private class EndTxnHandler extends TxnRequestHandler {
        private final EndTxnRequest.Builder builder;

        private EndTxnHandler(EndTxnRequest.Builder builder) {
            this.builder = builder;
        }

        @Override
        EndTxnRequest.Builder requestBuilder() {
            return builder;
        }

        @Override
        Priority priority() {
            return Priority.END_TXN;
        }

        @Override
        boolean isEndTxn() {
            return true;
        }

        @Override
        public void handleResponse(AbstractResponse response) {
            EndTxnResponse endTxnResponse = (EndTxnResponse) response;
            Errors error = endTxnResponse.error();

            if (error == Errors.NONE) {
                completeTransaction();
                result.done();
            } else if (error == Errors.COORDINATOR_NOT_AVAILABLE || error == Errors.NOT_COORDINATOR) {
                lookupCoordinator(FindCoordinatorRequest.CoordinatorType.TRANSACTION, transactionalId);
                reenqueue();
            } else if (error == Errors.COORDINATOR_LOAD_IN_PROGRESS || error == Errors.CONCURRENT_TRANSACTIONS) {
                reenqueue();
            } else if (error == Errors.INVALID_PRODUCER_EPOCH) {
                fatalError(error.exception());
            } else if (error == Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED) {
                fatalError(error.exception());
            } else if (error == Errors.INVALID_TXN_STATE) {
                fatalError(error.exception());
            } else {
                fatalError(new KafkaException("Unhandled error in EndTxnResponse: " + error.message()));
            }
        }
    }

    private class AddOffsetsToTxnHandler extends TxnRequestHandler {
        private final AddOffsetsToTxnRequest.Builder builder;
        private final Map<TopicPartition, OffsetAndMetadata> offsets;

        private AddOffsetsToTxnHandler(AddOffsetsToTxnRequest.Builder builder,
                                       Map<TopicPartition, OffsetAndMetadata> offsets) {
            this.builder = builder;
            this.offsets = offsets;
        }

        @Override
        AddOffsetsToTxnRequest.Builder requestBuilder() {
            return builder;
        }

        @Override
        Priority priority() {
            return Priority.ADD_PARTITIONS_OR_OFFSETS;
        }

        @Override
        public void handleResponse(AbstractResponse response) {
            AddOffsetsToTxnResponse addOffsetsToTxnResponse = (AddOffsetsToTxnResponse) response;
            Errors error = addOffsetsToTxnResponse.error();

            if (error == Errors.NONE) {
                log.debug("Successfully added partition for consumer group {} to transaction", builder.consumerGroupId());

                // note the result is not completed until the TxnOffsetCommit returns
                pendingRequests.add(txnOffsetCommitHandler(result, offsets, builder.consumerGroupId()));
                transactionStarted = true;
            } else if (error == Errors.COORDINATOR_NOT_AVAILABLE || error == Errors.NOT_COORDINATOR) {
                lookupCoordinator(FindCoordinatorRequest.CoordinatorType.TRANSACTION, transactionalId);
                reenqueue();
            } else if (error == Errors.COORDINATOR_LOAD_IN_PROGRESS || error == Errors.CONCURRENT_TRANSACTIONS) {
                reenqueue();
            } else if (error == Errors.INVALID_PRODUCER_EPOCH) {
                fatalError(error.exception());
            } else if (error == Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED) {
                fatalError(error.exception());
            } else if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
                abortableError(new GroupAuthorizationException(builder.consumerGroupId()));
            } else {
                fatalError(new KafkaException("Unexpected error in AddOffsetsToTxnResponse: " + error.message()));
            }
        }
    }

    private class TxnOffsetCommitHandler extends TxnRequestHandler {
        private final TxnOffsetCommitRequest.Builder builder;

        private TxnOffsetCommitHandler(TransactionalRequestResult result,
                                       TxnOffsetCommitRequest.Builder builder) {
            super(result);
            this.builder = builder;
        }

        @Override
        TxnOffsetCommitRequest.Builder requestBuilder() {
            return builder;
        }

        @Override
        Priority priority() {
            return Priority.ADD_PARTITIONS_OR_OFFSETS;
        }

        @Override
        FindCoordinatorRequest.CoordinatorType coordinatorType() {
            return FindCoordinatorRequest.CoordinatorType.GROUP;
        }

        @Override
        String coordinatorKey() {
            return builder.consumerGroupId();
        }

        @Override
        public void handleResponse(AbstractResponse response) {
            TxnOffsetCommitResponse txnOffsetCommitResponse = (TxnOffsetCommitResponse) response;
            boolean coordinatorReloaded = false;
            boolean hadFailure = false;
            Map<TopicPartition, Errors> errors = txnOffsetCommitResponse.errors();

            for (Map.Entry<TopicPartition, Errors> entry : errors.entrySet()) {
                TopicPartition topicPartition = entry.getKey();
                Errors error = entry.getValue();
                if (error == Errors.NONE) {
                    log.debug("Successfully added offsets {} from consumer group {} to transaction.",
                            builder.offsets(), builder.consumerGroupId());
                    pendingTxnOffsetCommits.remove(topicPartition);
                } else if (error == Errors.COORDINATOR_NOT_AVAILABLE
                        || error == Errors.NOT_COORDINATOR
                        || error == Errors.REQUEST_TIMED_OUT) {
                    hadFailure = true;
                    if (!coordinatorReloaded) {
                        coordinatorReloaded = true;
                        lookupCoordinator(FindCoordinatorRequest.CoordinatorType.GROUP, builder.consumerGroupId());
                    }
                } else if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
                    hadFailure = true;
                } else if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
                    abortableError(new GroupAuthorizationException(builder.consumerGroupId()));
                    return;
                } else if (error == Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED
                        || error == Errors.INVALID_PRODUCER_EPOCH
                        || error == Errors.UNSUPPORTED_FOR_MESSAGE_FORMAT) {
                    fatalError(error.exception());
                    return;
                } else {
                    fatalError(new KafkaException("Unexpected error in TxnOffsetCommitResponse: " + error.message()));
                    return;
                }
            }

            if (!hadFailure || !result.isSuccessful()) {
                // all attempted partitions were either successful, or there was a fatal failure.
                // either way, we are not retrying, so complete the request.
                result.done();
                return;
            }

            // retry the commits which failed with a retriable error.
            if (!pendingTxnOffsetCommits.isEmpty())
                reenqueue();
        }
    }
}


text:
@extends('layouts.default')

@section('content')

    <form class="form-horizontal"  method="post">
        {{ csrf_field() }}
        <fieldset>
            <legend>Nueva sugerencia</legend>

                Título

                    <input type="text" class="form-control" name="title" id="title" placeholder="Escribe un nuevo titulo">

                Expón el problema

                    <input type="text" class="form-control" name="body" id="body" placeholder="">

                    <button type="submit" class="btn btn-primary">Crear</button>

        </fieldset>
    </form>

@stop

redpajama_set_name: RedPajamaGithub


text:
package org.locationtech.geowave.test.docs;

import java.util.Date;
import org.geotools.feature.AttributeTypeBuilder;
import org.geotools.feature.simple.SimpleFeatureBuilder;
import org.geotools.feature.simple.SimpleFeatureTypeBuilder;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.locationtech.geowave.adapter.vector.FeatureDataAdapter;
import org.locationtech.geowave.core.geotime.index.api.SpatialIndexBuilder;
import org.locationtech.geowave.core.geotime.index.api.SpatialTemporalIndexBuilder;
import org.locationtech.geowave.core.geotime.index.dimension.TemporalBinningStrategy.Unit;
import org.locationtech.geowave.core.geotime.store.query.api.VectorAggregationQueryBuilder;
import org.locationtech.geowave.core.geotime.store.query.api.VectorQueryBuilder;
import org.locationtech.geowave.core.geotime.store.query.api.VectorQueryConstraintsFactory;
import org.locationtech.geowave.core.geotime.store.statistics.BoundingBoxStatistic;
import org.locationtech.geowave.core.geotime.store.statistics.BoundingBoxStatistic.BoundingBoxValue;
import org.locationtech.geowave.core.geotime.store.statistics.SpatialTemporalStatisticQueryBuilder;
import org.locationtech.geowave.core.index.persist.Persistable;
import org.locationtech.geowave.core.store.CloseableIterator;
import org.locationtech.geowave.core.store.StoreFactoryOptions;
import org.locationtech.geowave.core.store.api.AggregationQuery;
import org.locationtech.geowave.core.store.api.DataStore;
import org.locationtech.geowave.core.store.api.DataStoreFactory;
import org.locationtech.geowave.core.store.api.Index;
import org.locationtech.geowave.core.store.api.Query;
import org.locationtech.geowave.core.store.api.StatisticQuery;
import org.locationtech.geowave.core.store.api.StatisticQueryBuilder;
import org.locationtech.geowave.core.store.api.Writer;
import org.locationtech.geowave.core.store.cli.store.DataStorePluginOptions;
import org.locationtech.geowave.core.store.statistics.query.FieldStatisticQueryBuilder;
import org.locationtech.geowave.test.GeoWaveITRunner;
import org.locationtech.geowave.test.annotation.GeoWaveTestStore;
import org.locationtech.geowave.test.annotation.GeoWaveTestStore.GeoWaveStoreType;
import org.locationtech.geowave.test.basic.AbstractGeoWaveIT;
import org.locationtech.jts.geom.Coordinate;
import org.locationtech.jts.geom.Envelope;
import org.locationtech.jts.geom.GeometryFactory;
import org.locationtech.jts.geom.Point;
import org.opengis.feature.simple.SimpleFeature;
import org.opengis.feature.simple.SimpleFeatureType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@RunWith(GeoWaveITRunner.class)
public class GeoWaveDocumentationExamplesIT extends AbstractGeoWaveIT {
  private static final Logger LOGGER =
      LoggerFactory.getLogger(GeoWaveDocumentationExamplesIT.class);

  @GeoWaveTestStore(
      value = {
          GeoWaveStoreType.ACCUMULO,
          GeoWaveStoreType.BIGTABLE,
          GeoWaveStoreType.CASSANDRA,
          GeoWaveStoreType.DYNAMODB,
          GeoWaveStoreType.HBASE,
          GeoWaveStoreType.KUDU,
          GeoWaveStoreType.REDIS,
          GeoWaveStoreType.ROCKSDB,
          GeoWaveStoreType.FILESYSTEM})
  protected DataStorePluginOptions dataStore;

  private static long startMillis;

  @BeforeClass
  public static void reportTestStart() {
    startMillis = System.currentTimeMillis();
    LOGGER.warn("------------------------------------------");
    LOGGER.warn("*                                        *");
    LOGGER.warn("* RUNNING GeoWaveDocumentationExamplesIT *");
    LOGGER.warn("*                                        *");
    LOGGER.warn("------------------------------------------");
  }

  @AfterClass
  public static void reportTestFinish() {
    LOGGER.warn("-------------------------------------------");
    LOGGER.warn("*                                         *");
    LOGGER.warn("* FINISHED GeoWaveDocumentationExamplesIT *");
    LOGGER.warn(
        "*                "
            + ((System.currentTimeMillis() - startMillis) / 1000)
            + "s elapsed.             *");
    LOGGER.warn("*                                         *");
    LOGGER.warn("-------------------------------------------");
  }

  @Test
  public void testExamples() throws Exception {
    // !!IMPORTANT!! If this test has to be updated, update the associated programmatic API example
    // in the dev guide!
    StoreFactoryOptions options = dataStore.getFactoryOptions();

    DataStore myStore = DataStoreFactory.createDataStore(options);

    // --------------------------------------------------------------------
    // Create Indices Example !! See Note at Top of Test
    // --------------------------------------------------------------------
    // Spatial Index
    SpatialIndexBuilder spatialIndexBuilder = new SpatialIndexBuilder();
    spatialIndexBuilder.setCrs("EPSG:4326");
    Index spatialIndex = spatialIndexBuilder.createIndex();

    // Spatial-temporal Index
    SpatialTemporalIndexBuilder spatialTemporalIndexBuilder = new SpatialTemporalIndexBuilder();
    spatialTemporalIndexBuilder.setCrs("EPSG:3857");
    spatialTemporalIndexBuilder.setPeriodicity(Unit.MONTH);
    Index spatialTemporalIndex = spatialTemporalIndexBuilder.createIndex();
    // --------------------------------------------------------------------

    // --------------------------------------------------------------------
    // Add Indices Example !! See Note at Top of Test
    // --------------------------------------------------------------------
    // Add the spatial and spatial-temporal indices
    myStore.addIndex(spatialIndex);
    myStore.addIndex(spatialTemporalIndex);
    // --------------------------------------------------------------------

    // --------------------------------------------------------------------
    // Ingest Example !! See Note at Top of Test
    // --------------------------------------------------------------------
    // Create a point feature type
    SimpleFeatureTypeBuilder pointTypeBuilder = new SimpleFeatureTypeBuilder();
    AttributeTypeBuilder attributeBuilder = new AttributeTypeBuilder();
    pointTypeBuilder.setName("TestPointType");
    pointTypeBuilder.add(
        attributeBuilder.binding(Point.class).nillable(false).buildDescriptor("the_geom"));
    pointTypeBuilder.add(
        attributeBuilder.binding(Date.class).nillable(false).buildDescriptor("date"));
    SimpleFeatureType pointType = pointTypeBuilder.buildFeatureType();

    // Create a feature builder
    SimpleFeatureBuilder pointFeatureBuilder = new SimpleFeatureBuilder(pointType);

    // Create an adapter for point type
    FeatureDataAdapter pointTypeAdapter = new FeatureDataAdapter(pointType);

    // Add the point type to the data store in the spatial index
    myStore.addType(pointTypeAdapter, spatialIndex);

    // Create a writer to ingest data
    try (Writer<SimpleFeature> writer = myStore.createWriter(pointTypeAdapter.getTypeName())) {
      // Write some features to the data store
      GeometryFactory factory = new GeometryFactory();
      pointFeatureBuilder.set("the_geom", factory.createPoint(new Coordinate(1, 1)));
      pointFeatureBuilder.set("date", new Date());
      writer.write(pointFeatureBuilder.buildFeature("feature1"));

      pointFeatureBuilder.set("the_geom", factory.createPoint(new Coordinate(5, 5)));
      pointFeatureBuilder.set("date", new Date());
      writer.write(pointFeatureBuilder.buildFeature("feature2"));

      pointFeatureBuilder.set("the_geom", factory.createPoint(new Coordinate(-5, -5)));
      pointFeatureBuilder.set("date", new Date());
      writer.write(pointFeatureBuilder.buildFeature("feature3"));
    }
    // --------------------------------------------------------------------

    // --------------------------------------------------------------------
    // Query Data Example !! See Note at Top of Test
    // --------------------------------------------------------------------
    // Create the query builder and constraints factory
    VectorQueryBuilder queryBuilder = VectorQueryBuilder.newBuilder();
    VectorQueryConstraintsFactory constraintsFactory = queryBuilder.constraintsFactory();

    // Use the constraints factory to create a bounding box constraint
    queryBuilder.constraints(constraintsFactory.cqlConstraints("BBOX(the_geom, -1, -1, 6, 6)"));

    // Only query data from the point type
    queryBuilder.addTypeName(pointTypeAdapter.getTypeName());

    // Build the query
    Query<SimpleFeature> query = queryBuilder.build();

    // Execute the query
    try (CloseableIterator<SimpleFeature> features = myStore.query(query)) {
      // Iterate through the results
      while (features.hasNext()) {
        SimpleFeature feature = features.next();
        // Do something with the feature
      }
    }
    // --------------------------------------------------------------------
    // Verify example
    try (CloseableIterator<SimpleFeature> features = myStore.query(queryBuilder.build())) {
      // Iterate through the results
      int featureCount = 0;
      while (features.hasNext()) {
        features.next();
        featureCount++;
        // Do something with the feature
      }
      Assert.assertEquals(2, featureCount);
    }

    // --------------------------------------------------------------------
    // Aggregation Example !! See Note at Top of Test
    // --------------------------------------------------------------------
    // Create the aggregation query builder
    VectorAggregationQueryBuilder<Persistable, Object> aggregationQueryBuilder =
        VectorAggregationQueryBuilder.newBuilder();

    // Use the constraints factory from the previous example to create a bounding box constraint
    aggregationQueryBuilder.constraints(
        constraintsFactory.cqlConstraints("BBOX(the_geom, -1, -1, 6, 6)"));

    // Configure the query to use a count aggregation on the desired type
    aggregationQueryBuilder.count(pointTypeAdapter.getTypeName());

    // Create the aggregation query
    AggregationQuery<Persistable, Object, SimpleFeature> aggregationQuery =
        aggregationQueryBuilder.build();

    // Perform the aggregation
    long count = (Long) myStore.aggregate(aggregationQuery);
    // --------------------------------------------------------------------
    // Verify example
    Assert.assertEquals(2, count);

    // --------------------------------------------------------------------
    // Statistics Example !! See Note at Top of Test
    // --------------------------------------------------------------------
    // Create the statistics query builder for the BoundingBoxStatistic
    FieldStatisticQueryBuilder<BoundingBoxValue, Envelope> builder =
        SpatialTemporalStatisticQueryBuilder.bbox();

    // Specify the type name
    builder.typeName(pointTypeAdapter.getTypeName());

    // Create the bounding box statistics query
    StatisticQuery<BoundingBoxValue, Envelope> bboxQuery = builder.build();

    // Aggregate the statistic into a single result
    BoundingBoxValue bboxStatValue = myStore.aggregateStatistics(bboxQuery);

    // Get the value
    Envelope bbox = bboxStatValue.getValue();
    // --------------------------------------------------------------------
    // Verify example
    Assert.assertEquals(-5.0, bbox.getMinX(), 0.0001);
    Assert.assertEquals(-5.0, bbox.getMinY(), 0.0001);
    Assert.assertEquals(5.0, bbox.getMaxX(), 0.0001);
    Assert.assertEquals(5.0, bbox.getMaxY(), 0.0001);
  }

  @Override
  protected DataStorePluginOptions getDataStorePluginOptions() {
    return dataStore;
  }
}


text:
package org.apache.jasper.tagplugins.jstl.core;

import javax.servlet.http.HttpServletResponse;

import org.junit.Assert;
import org.junit.Test;

import org.apache.tomcat.util.buf.ByteChunk;

public class TestForEach extends AbstractTestTag {

    @Test
    public void testBug54242() throws Exception {
        ByteChunk res = new ByteChunk();

        int rc = getUrl("http://localhost:" + getPort() +
                "/test/bug5nnnn/bug54242.jsp", res, null);

        Assert.assertEquals(HttpServletResponse.SC_OK, rc);

        String body = res.toString();
        Assert.assertTrue(body.contains("OK - 1"));
        Assert.assertTrue(body.contains("OK - 2"));
        Assert.assertFalse(body.contains("FAIL"));
    }

    @Test
    public void testBug54888() throws Exception {
        ByteChunk res = new ByteChunk();

        int rc = getUrl("http://localhost:" + getPort() +
                "/test/bug5nnnn/bug54888.jsp", res, null);

        Assert.assertEquals(HttpServletResponse.SC_OK, rc);

        String body = res.toString();
        Assert.assertTrue(body.contains("OK - 1"));
        Assert.assertTrue(body.contains("OK - 2"));
        Assert.assertTrue(body.contains("OK - 3"));
    }
}


text:
@echo off
setlocal

echo =================================
echo Making assembly manifest for TradeWright.Common.ExternalComponents
echo.

%TW-PROJECTS-DRIVE%
path %TW-PROJECTS-DRIVE%%TW-PROJECTS-PATH%\Build\Subscripts;%PATH%

call setTradeWrightCommonVersion.bat

pushd %TW-PROJECTS-PATH%\Build
generateManifest /Ass:TradeWright.Common.ExternalComponents,%VB6-BUILD-MAJOR%.%VB6-BUILD-MINOR%.0.%VB6-BUILD-REVISION%,"TradeWright Common ExternalComponents",TradeWrightCommonExternalComponents.txt ^
                 /Out:..\Bin\TradeWright.Common.ExternalComponents\TradeWright.Common.ExternalComponents.manifest ^
                 /Inline
if errorlevel 1 goto :err

echo Manifest generated
popd
exit /B

:err
echo Manifest generation failed
popd
exit /B 1


text:
package com.theenginerd.ccscada.block.factory

import cpw.mods.fml.common.registry.GameRegistry
import com.theenginerd.ccscada.block.RedstoneControllerBlock
import com.theenginerd.ccscada.tileentity.PeripheralTileEntity
import net.minecraft.tileentity.TileEntity
import net.minecraft.world.World
import com.theenginerd.ccscada.peripheral.RedstoneController

object RedstoneControllerBlockFactory extends BlockFactory
{
    def createBlock(blockId: Int) =
    {
        class RedstoneControllerTileEntity
            extends PeripheralTileEntity
            with RedstoneController

        GameRegistry.registerTileEntity(classOf[RedstoneControllerTileEntity], "redstoneControllerPeripheral")

        new RedstoneControllerBlock(blockId)
        {
            override def createNewTileEntity(world: World): TileEntity =
            {
                new RedstoneControllerTileEntity
            }
        }
    }
}


text:
const FILE_PATTERN = /.+\.(json|js)$/;
const LOCALE_PATTERN = /.+\/(\w+)\.(json|js)$/;

var path = require('path');

//var co = require('co');
var exists = require('fs').existsSync;

var JsonLoader = function JsonLoader(file, locale, msg) {
  var _messages = msg;

  Object.defineProperties(this, {
    locale: {
      configurable: false,
      enumerable: true,
      writable: false,
      value: locale,
    },
    source: {
      configurable: false,
      enumerable: true,
      writable: false,
      value: file,
    },
    contains: {
      configurable: false,
      enumerable: true,
      writable: false,
      value: function * containsMessage(msgId) {
        return !!_messages[msgId];
      }
    },
    get: {
      configurable: false,
      enumerable: true,
      writable: false,
      value: function * getMessage(msgId) {
        return _messages[msgId];
      }
    },
    forEach: {   // TODO : replace this with __iterator__ once it is implemented
      configurable: false,
      enumerable: true,
      writable: false,
      value: function forEach(iterator) {
        for (var msgId in _messages) {
          if (iterator(msgId) === false) {
            break;
          }
        }
      }
    }
  });
};

/**
Load the given JSON or JS module and return it. The module is supposed to be
an object whose keys and values are I18N translation strings. The file argument
should be an absolute file name.
*/
module.exports = function * jsonLoader(file) {
  var locale;
  var messages;

  if (typeof file !== 'string' || !FILE_PATTERN.test(file) || !exists(file)) {
    return false;
  }

  locale = file.replace(LOCALE_PATTERN, '$1');
  messages = require(path.resolve(process.cwd(), file));

  return new JsonLoader(file, locale, messages);
};


text:
enum Choice { COOPERATE, RETALIATE };

class Dilemma
{
	public:
		Dilemma() { }
		virtual ~Dilemma() { }

	public:
		virtual Choice SendMyMove() = 0;

	public:
		void RecordMyMove(Choice myMove)
		{
			_myMoves.push_back(myMove);
		}

		void RecordTheirMove(Choice theirMove)
		{
			_theirMoves.push_back(theirMove);
		}

	protected:
		std::vector<Choice> _myMoves;
		std::vector<Choice> _theirMoves;
};

#endif


text:
#ifndef MQTTMESSAGINGSTUB_H
#define MQTTMESSAGINGSTUB_H

#include <functional>
#include <memory>

#include "joynr/IMessagingStub.h"
#include "joynr/Logger.h"
#include "joynr/PrivateCopyAssign.h"
#include "joynr/system/RoutingTypes/MqttAddress.h"

namespace joynr
{

class ImmutableMessage;
class ITransportMessageSender;

namespace exceptions
{
class JoynrRuntimeException;
}

/**
 * Is used by the ClusterController to contact another (remote) ClusterController
 */
class MqttMessagingStub : public IMessagingStub
{
public:
    explicit MqttMessagingStub(std::shared_ptr<ITransportMessageSender> messageSender,
                               const system::RoutingTypes::MqttAddress& destinationAddress);
    ~MqttMessagingStub() override = default;
    void transmit(std::shared_ptr<ImmutableMessage> message,
                  const std::function<void(const exceptions::JoynrRuntimeException&)>& onFailure)
            override;

private:
    DISALLOW_COPY_AND_ASSIGN(MqttMessagingStub);
    std::shared_ptr<ITransportMessageSender> _messageSender;
    const system::RoutingTypes::MqttAddress _destinationAddress;
    ADD_LOGGER(MqttMessagingStub)
};

} // namespace joynr
#endif // MQTTMESSAGINGSTUB_H


text:
import distutils.dist
from distutils.util \
    import \
        check_environ, strtobool, rfc822_escape

# Taken from python 2.6.5
PKG_INFO_ENCODING = 'utf-8'
class _DistributionMetadata:
    """Dummy class to hold the distribution meta-data: name, version,
    author, and so forth.
    """

    _METHOD_BASENAMES = ("name", "version", "author", "author_email",
                         "maintainer", "maintainer_email", "url",
                         "license", "description", "long_description",
                         "keywords", "platforms", "fullname", "contact",
                         "contact_email", "license", "classifiers",
                         "download_url",
                         # PEP 314
                         "provides", "requires", "obsoletes",
                         )

    def __init__ (self):
        self.name = None
        self.version = None
        self.author = None
        self.author_email = None
        self.maintainer = None
        self.maintainer_email = None
        self.url = None
        self.license = None
        self.description = None
        self.long_description = None
        self.keywords = None
        self.platforms = None
        self.classifiers = None
        self.download_url = None
        # PEP 314
        self.provides = None
        self.requires = None
        self.obsoletes = None

    def write_pkg_info (self, base_dir):
        """Write the PKG-INFO file into the release tree.
        """
        pkg_info = open( os.path.join(base_dir, 'PKG-INFO'), 'w')

        self.write_pkg_file(pkg_info)

        pkg_info.close()

    # write_pkg_info ()

    def write_pkg_file (self, file):
        """Write the PKG-INFO format data to a file object.
        """
        version = '1.0'
        if self.provides or self.requires or self.obsoletes:
            version = '1.1'

        self._write_field(file, 'Metadata-Version', version)
        self._write_field(file, 'Name', self.get_name())
        self._write_field(file, 'Version', self.get_version())
        self._write_field(file, 'Summary', self.get_description())
        self._write_field(file, 'Home-page', self.get_url())
        self._write_field(file, 'Author', self.get_contact())
        self._write_field(file, 'Author-email', self.get_contact_email())
        self._write_field(file, 'License', self.get_license())
        if self.download_url:
            self._write_field(file, 'Download-URL', self.download_url)

        long_desc = rfc822_escape( self.get_long_description())
        self._write_field(file, 'Description', long_desc)

        keywords = ",".join(self.get_keywords())
        if keywords:
            self._write_field(file, 'Keywords', keywords)

        self._write_list(file, 'Platform', self.get_platforms())
        self._write_list(file, 'Classifier', self.get_classifiers())

        # PEP 314
        self._write_list(file, 'Requires', self.get_requires())
        self._write_list(file, 'Provides', self.get_provides())
        self._write_list(file, 'Obsoletes', self.get_obsoletes())

    def _write_field(self, file, name, value):
        file.write('%s: %s\n' % (name, self._encode_field(value)))

    def _write_list (self, file, name, values):

        for value in values:
            self._write_field(file, name, value)

    def _encode_field(self, value):
        if value is None:
            return None
        if isinstance(value, unicode):
            return value.encode(PKG_INFO_ENCODING)
        return str(value)

    # -- Metadata query methods ----------------------------------------

    def get_name (self):
        return self.name or "UNKNOWN"

    def get_version(self):
        return self.version or "0.0.0"

    def get_fullname (self):
        return "%s-%s" % (self.get_name(), self.get_version())

    def get_author(self):
        return self._encode_field(self.author) or "UNKNOWN"

    def get_author_email(self):
        return self.author_email or "UNKNOWN"

    def get_maintainer(self):
        return self._encode_field(self.maintainer) or "UNKNOWN"

    def get_maintainer_email(self):
        return self.maintainer_email or "UNKNOWN"

    def get_contact(self):
        return (self._encode_field(self.maintainer) or
                self._encode_field(self.author) or "UNKNOWN")

    def get_contact_email(self):
        return (self.maintainer_email or
                self.author_email or
                "UNKNOWN")

    def get_url(self):
        return self.url or "UNKNOWN"

    def get_license(self):
        return self.license or "UNKNOWN"
    get_licence = get_license

    def get_description(self):
        return self._encode_field(self.description) or "UNKNOWN"

    def get_long_description(self):
        return self._encode_field(self.long_description) or "UNKNOWN"

    def get_keywords(self):
        return self.keywords or []

    def get_platforms(self):
        return self.platforms or ["UNKNOWN"]

    def get_classifiers(self):
        return self.classifiers or []

    def get_download_url(self):
        return self.download_url or "UNKNOWN"

    # PEP 314

    def get_requires(self):
        return self.requires or []

    def set_requires(self, value):
        import distutils.versionpredicate
        for v in value:
            distutils.versionpredicate.VersionPredicate(v)
        self.requires = value

    def get_provides(self):
        return self.provides or []

    def set_provides(self, value):
        value = [v.strip() for v in value]
        for v in value:
            import distutils.versionpredicate
            distutils.versionpredicate.split_provision(v)
        self.provides = value

    def get_obsoletes(self):
        return self.obsoletes or []

    def set_obsoletes(self, value):
        import distutils.versionpredicate
        for v in value:
            distutils.versionpredicate.VersionPredicate(v)
        self.obsoletes = value

if not hasattr(distutils.dist.DistributionMetadata, "write_pkg_file"):
    DistributionMetadata = _DistributionMetadata
else:
    DistributionMetadata = distutils.dist.DistributionMetadata


text:
#include <aws/databrew/model/CreateProfileJobRequest.h>
#include <aws/core/utils/json/JsonSerializer.h>

#include <utility>

using namespace Aws::GlueDataBrew::Model;
using namespace Aws::Utils::Json;
using namespace Aws::Utils;

CreateProfileJobRequest::CreateProfileJobRequest() :
    m_datasetNameHasBeenSet(false),
    m_encryptionKeyArnHasBeenSet(false),
    m_encryptionMode(EncryptionMode::NOT_SET),
    m_encryptionModeHasBeenSet(false),
    m_nameHasBeenSet(false),
    m_logSubscription(LogSubscription::NOT_SET),
    m_logSubscriptionHasBeenSet(false),
    m_maxCapacity(0),
    m_maxCapacityHasBeenSet(false),
    m_maxRetries(0),
    m_maxRetriesHasBeenSet(false),
    m_outputLocationHasBeenSet(false),
    m_configurationHasBeenSet(false),
    m_roleArnHasBeenSet(false),
    m_tagsHasBeenSet(false),
    m_timeout(0),
    m_timeoutHasBeenSet(false),
    m_jobSampleHasBeenSet(false)
{
}

Aws::String CreateProfileJobRequest::SerializePayload() const
{
  JsonValue payload;

  if(m_datasetNameHasBeenSet)
  {
   payload.WithString("DatasetName", m_datasetName);

  }

  if(m_encryptionKeyArnHasBeenSet)
  {
   payload.WithString("EncryptionKeyArn", m_encryptionKeyArn);

  }

  if(m_encryptionModeHasBeenSet)
  {
   payload.WithString("EncryptionMode", EncryptionModeMapper::GetNameForEncryptionMode(m_encryptionMode));
  }

  if(m_nameHasBeenSet)
  {
   payload.WithString("Name", m_name);

  }

  if(m_logSubscriptionHasBeenSet)
  {
   payload.WithString("LogSubscription", LogSubscriptionMapper::GetNameForLogSubscription(m_logSubscription));
  }

  if(m_maxCapacityHasBeenSet)
  {
   payload.WithInteger("MaxCapacity", m_maxCapacity);

  }

  if(m_maxRetriesHasBeenSet)
  {
   payload.WithInteger("MaxRetries", m_maxRetries);

  }

  if(m_outputLocationHasBeenSet)
  {
   payload.WithObject("OutputLocation", m_outputLocation.Jsonize());

  }

  if(m_configurationHasBeenSet)
  {
   payload.WithObject("Configuration", m_configuration.Jsonize());

  }

  if(m_roleArnHasBeenSet)
  {
   payload.WithString("RoleArn", m_roleArn);

  }

  if(m_tagsHasBeenSet)
  {
   JsonValue tagsJsonMap;
   for(auto& tagsItem : m_tags)
   {
     tagsJsonMap.WithString(tagsItem.first, tagsItem.second);
   }
   payload.WithObject("Tags", std::move(tagsJsonMap));

  }

  if(m_timeoutHasBeenSet)
  {
   payload.WithInteger("Timeout", m_timeout);

  }

  if(m_jobSampleHasBeenSet)
  {
   payload.WithObject("JobSample", m_jobSample.Jsonize());

  }

  return payload.View().WriteReadable();
}


text:
\hypertarget{_glossy_specular_8h}{}\section{Source/\+B\+R\+D\+Fs/include/\+Glossy\+Specular.h File Reference}
\label{_glossy_specular_8h}\index{Source/\+B\+R\+D\+Fs/include/\+Glossy\+Specular.\+h@{Source/\+B\+R\+D\+Fs/include/\+Glossy\+Specular.\+h}}
{\ttfamily \#include \char`\"{}B\+R\+D\+F.\+h\char`\"{}}\newline
\subsection*{Classes}
\begin{DoxyCompactItemize}
\item
class \hyperlink{class_glossy_specular}{Glossy\+Specular}
\end{DoxyCompactItemize}


text:
package com.mjrichardson.teamCity.buildTriggers;

import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;

public class OctopusDate {
    private static final String OCTOPUS_DATE_FORMAT = "yyyy-MM-ddHH:mm:ss.SSSZ";//2015-12-08T08:09:39.624+00:00
    private DateTime dateTime;

    private OctopusDate(DateTime dateTime) {
        this.dateTime = dateTime;
    }

    public OctopusDate(int year, int month, int dayOfMonth) {
        this(year, month, dayOfMonth, 0, 0, 0, 0);
    }

    public OctopusDate(int year, int month, int dayOfMonth, int hour, int minute, int seconds) {
        this(year, month, dayOfMonth, hour, minute, seconds, 0);
    }

    public OctopusDate(int year, int month, int dayOfMonth, int hour, int minute, int seconds, int milliseconds) {
        dateTime = new DateTime(year, month, dayOfMonth, hour, minute, seconds, milliseconds, DateTimeZone.UTC);
    }

    public static OctopusDate Parse(String input) {
        DateTimeFormatter dateFormat = DateTimeFormat.forPattern(OCTOPUS_DATE_FORMAT);
        DateTime dateTime = DateTime.parse(input.replace("Z", "+00:00").replace("T", ""), dateFormat);
        if (dateTime.compareTo(new DateTime(1970, 1, 1, 0, 0, 0, 0, DateTimeZone.UTC)) == 0)
            return new NullOctopusDate();
        return new OctopusDate(dateTime);
    }

    public int compareTo(OctopusDate compareDate) {
        return dateTime.compareTo(compareDate.dateTime);
    }

    @Override
    public boolean equals(Object obj) {
        if (obj.getClass() != OctopusDate.class && obj.getClass() != NullOctopusDate.class)
            return false;
        return obj.toString().equals(toString());
    }

    @Override
    public String toString() {
        DateTimeFormatter dateFormat = DateTimeFormat.forPattern("yyyy-MM-dd'T'HH:mm:ss.SSSZZ").withZoneUTC();
        return dateFormat.print(dateTime);
    }
}


text:
// Copyright 2021 The Chromium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "chromeos/crosapi/cpp/gurl_os_handler_utils.h"

#include <stddef.h>

#include "base/logging.h"
#include "base/strings/string_util.h"
#include "url/url_util.h"

namespace {

const char kOsScheme[] = "os";
const char kOsUrlPrefix[] = "os://";
const char kChromeUIScheme[] = "chrome";
const char kChromeUrlPrefix[] = "chrome://";
const char kOsUISettingsURL[] = "os://settings";
const char kChromeUIOSSettingsHost[] = "os-settings";

// The start of the host portion of a GURL which starts with the os scheme.
const size_t kHostStart = sizeof(kOsUrlPrefix) - 1;

// Used for sanitation - any of the characters will cut the rest of the URL.
const char kTerminatingCharacters[] = "/\\? #.%$&*<>+";

// Note that GURL can't operate on the "os://" scheme as it is intentionally
// not a registered scheme.
std::string GetValidHostAndSubhostFromOsUrl(GURL url, bool include_path) {
  // Only keep the scheme, host and sub-host. Everything else gets cut off.
  const std::string& url_spec = base::ToLowerASCII(url.spec());

  // Find the first character after the host start
  std::size_t valid_spec_end =
      url_spec.find_first_of(kTerminatingCharacters, kHostStart);

  if (valid_spec_end == std::string::npos)
    return url_spec.substr(kHostStart);

  if (url_spec[valid_spec_end] == '/' && include_path) {
    // A sub URL is allowed (e.g. chrome://settings/network) - so we skip the
    // "/" with +1 and parse till we find the next terminating character.
    const std::size_t sub_host_end =
        url_spec.find_first_of(kTerminatingCharacters, valid_spec_end + 1);

    if (sub_host_end == std::string::npos)
      return url_spec.substr(kHostStart);

    if (sub_host_end > valid_spec_end + 1)
      valid_spec_end = sub_host_end;
  }

  // Copy beginning from after "os://" all characters in host and sub-host.
  return url_spec.substr(kHostStart, valid_spec_end - kHostStart);
}

GURL GetValidHostAndSubhostFromGURL(GURL gurl, bool include_path) {
  if (!gurl.is_valid() || !gurl.has_host())
    return GURL();

  if (!gurl.has_ref() && !gurl.has_username() && !gurl.has_password() &&
      !gurl.has_query() && !gurl.has_port() &&
      (include_path || !gurl.has_path())) {
    return GURL(gurl);
  }

  GURL::Replacements replacements;
  if (!include_path)
    replacements.ClearPath();
  replacements.ClearRef();
  replacements.ClearUsername();
  replacements.ClearPassword();
  replacements.ClearQuery();
  replacements.ClearPort();
  return gurl.ReplaceComponents(replacements);
}

}  // namespace

namespace crosapi {

namespace gurl_os_handler_utils {

GURL SanitizeAshURL(const GURL& url, bool include_path) {
  if (!IsAshOsUrl(url))
    return GetValidHostAndSubhostFromGURL(url, include_path);

  return GURL(kOsUrlPrefix +
              GetValidHostAndSubhostFromOsUrl(url, include_path));
}

GURL GetTargetURLFromLacrosURL(const GURL& url) {
  GURL target_url = crosapi::gurl_os_handler_utils::SanitizeAshURL(url);
  GURL short_target_url = crosapi::gurl_os_handler_utils::SanitizeAshURL(
      url, /*include_path=*/false);

  if (short_target_url != GURL(kOsUISettingsURL))
    return target_url;
  // Change os://settings/* into chrome://os-settings/* which will be the long
  // term home for our OS-settings.

  // This converts the os (GURL lib unusable) address into a chrome
  // (GURL lib usable) address.
  target_url =
      crosapi::gurl_os_handler_utils::GetChromeUrlFromSystemUrl(target_url);
  GURL::Replacements replacements;
  replacements.SetHostStr(kChromeUIOSSettingsHost);
  return target_url.ReplaceComponents(replacements);
}

bool IsUrlInList(const GURL& test_url, std::vector<GURL> list) {
  // It is assumed that the provided URL is sanitized as requested by
  // security at this point.
  DCHECK(SanitizeAshURL(test_url) == test_url);

  const GURL short_url = SanitizeAshURL(test_url, /*include_path=*/false);
  for (const GURL& url : list) {
    // It is expected that all os:// scheme items in the list are lower case
    // no "/" at the end and properly sanitized as the GURL comparison will
    // treat everything past the unknown scheme as an unknown string and hence
    // do no processing.
    DCHECK(!IsAshOsUrl(url) || SanitizeAshURL(url) == url);
    if (test_url == url || short_url == url)
      return true;
  }
  return false;
}

bool IsAshOsUrl(const GURL& url) {
  if (!url.is_valid() || url.spec().length() <= strlen(kOsUrlPrefix))
    return false;

  // Do not use "url.SchemeIs(kOsScheme)" here as it would require further
  // parsing of the "://" portion to see if the Url is also valid.
  return base::StartsWith(url.spec(), kOsUrlPrefix,
                          base::CompareCase::INSENSITIVE_ASCII);
}

bool IsAshOsAsciiScheme(const base::StringPiece& scheme) {
  return base::LowerCaseEqualsASCII(scheme, kOsScheme);
}

std::string AshOsUrlHost(const GURL& url) {
  if (!url.is_valid())
    return "";

  // If we are using a default scheme, GURL does all for us.
  if (!IsAshOsUrl(url)) {
    if (!url.has_host())
      return "";
    return url.host() + (url.path().length() > 1 ? url.path() : "");
  }

  return GetValidHostAndSubhostFromOsUrl(url, /*include_path=*/true);
}

// Convert a passed GURL from os:// to chrome://.
GURL GetSystemUrlFromChromeUrl(const GURL& url) {
  DCHECK(url.SchemeIs(kChromeUIScheme));
  return GURL(kOsUrlPrefix + url.host());
}

// Convert a passed GURL from chrome:// to os://.
GURL GetChromeUrlFromSystemUrl(const GURL& url) {
  DCHECK(IsAshOsUrl(url));
  return GURL(kChromeUrlPrefix + AshOsUrlHost(url));
}

}  // namespace gurl_os_handler_utils

}  // namespace crosapi


text:
<?php

namespace Application\Projection\Todo;

use Application\Projection\Table;
use Doctrine\DBAL\Connection;

/**
 * Class TodoFinder
 *
 * @package Application\Projection\Todo
 * @author Alexander Miertsch <kontakt@codeliner.ws>
 */
final class TodoFinder
{
    /**
     * @var Connection
     */
    private $connection;

    /**
     * @param Connection $connection
     */
    public function __construct(Connection $connection)
    {
        $this->connection = $connection;
    }

    /**
     * @return \stdClass[] of todoData
     */
    public function findAll()
    {
        return $this->connection->fetchAll(sprintf("SELECT * FROM %s", Table::TODO));
    }

    /**
     * @param string $assigneeId
     * @return \stdClass[] of todoData
     */
    public function findByAssigneeId($assigneeId)
    {
        return $this->connection->fetchAll(
            sprintf("SELECT * FROM %s WHERE assignee_id = :assignee_id", Table::TODO),
            ['assignee_id' => $assigneeId]
        );
    }

    /**
     * @param string $todoId
     * @return \stdClass of todoData
     */
    public function findById($todoId)
    {
        $stmt = $this->connection->prepare(sprintf("SELECT * FROM %s where id = :todo_id", Table::TODO));
        $stmt->bindValue('todo_id', $todoId);
        $stmt->execute();
        return $stmt->fetch();
    }
}


text:
"""
Django settings for tango_with_django_project project.

For more information on this file, see
https://docs.djangoproject.com/en/1.7/topics/settings/

For the full list of settings and their values, see
https://docs.djangoproject.com/en/1.7/ref/settings/
"""

# Build paths inside the project like this: os.path.join(BASE_DIR, ...)
import os
BASE_DIR = os.path.dirname(os.path.dirname(__file__))

# Quick-start development settings - unsuitable for production
# See https://docs.djangoproject.com/en/1.7/howto/deployment/checklist/

# SECURITY WARNING: keep the secret key used in production secret!
SECRET_KEY = 'k&en=nkp4c(bz*_h2h15a!c!jf&uiua-fca3_)9l4y7^!(o&fl'

# SECURITY WARNING: don't run with debug turned on in production!
DEBUG = False

TEMPLATE_DEBUG = True

TEMPLATE_PATH = os.path.join(BASE_DIR, 'templates')

TEMPLATE_DIRS = [ TEMPLATE_PATH, ]

ALLOWED_HOSTS = ['*']

# Application definition

INSTALLED_APPS = (
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'rango',
	'registration',
)

MIDDLEWARE_CLASSES = (
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.auth.middleware.SessionAuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
)

ROOT_URLCONF = 'tango_with_django_project.urls'

WSGI_APPLICATION = 'tango_with_django_project.wsgi.application'

# Database
# https://docs.djangoproject.com/en/1.7/ref/settings/#databases

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.sqlite3',
        'NAME': os.path.join(BASE_DIR, 'db.sqlite3'),
    }
}

# Internationalization
# https://docs.djangoproject.com/en/1.7/topics/i18n/

LANGUAGE_CODE = 'en-us'

TIME_ZONE = 'UTC'

USE_I18N = True

USE_L10N = True

USE_TZ = True

# Static files (CSS, JavaScript, Images)
# https://docs.djangoproject.com/en/1.7/howto/static-files/

STATIC_PATH = os.path.join(BASE_DIR,'static')

STATIC_URL = '/static/'

STATICFILES_DIRS = (
    STATIC_PATH,
)

MEDIA_URL = '/media/'
MEDIA_ROOT = os.path.join(BASE_DIR, 'media') # Absolute path to the media directory

PASSWORD_HASHERS = (
'django.contrib.auth.hashers.PBKDF2PasswordHasher',
'django.contrib.auth.hashers.PBKDF2SHA1PasswordHasher',
)

# LOGIN_URL = '/rango/login/'

REGISTRATION_OPEN = True        # If True, users can register
ACCOUNT_ACTIVATION_DAYS = 7     # One-week activation window; you may, of course, use a different value.
REGISTRATION_AUTO_LOGIN = True  # If True, the user will be automatically logged in.
LOGIN_REDIRECT_URL = '/rango/'  # The page you want users to arrive at after they successful log in
LOGIN_URL = '/accounts/login/'  # The page users are directed to if they are not logged in, and are trying to access pages requiring authentication