import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
/**
 * Get the next batch of records by polling.
 * @return Next batch of records or null if no records available.
 */
private ConsumerRecords<byte[], byte[]> pollRequests() {
    ConsumerRecords<byte[], byte[]> records = null;

    try {
        records = consumer.poll(pollTimeMs);
    } catch (final InvalidOffsetException e) {
        resetInvalidOffsets(e);
    }

    if (rebalanceException != null) {
        if (!(rebalanceException instanceof ProducerFencedException)) {
            throw new StreamsException(logPrefix + " Failed to rebalance.", rebalanceException);
        }
    }

    return records;
}
 

import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
private RuntimeException performOnStreamTasks(final StreamTaskAction action) {
    RuntimeException firstException = null;
    final Iterator<Map.Entry<TaskId, StreamTask>> it = activeTasks.entrySet().iterator();
    while (it.hasNext()) {
        final StreamTask task = it.next().getValue();
        try {
            action.apply(task);
        } catch (final ProducerFencedException e) {
            closeZombieTask(task);
            it.remove();
        } catch (final RuntimeException t) {
            log.error("{} Failed to {} stream task {}: ",
                logPrefix,
                action.name(),
                task.id(),
                t);
            if (firstException == null) {
                firstException = t;
            }
        }
    }

    return firstException;
}
 

import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
private void failBatch(ProducerBatch batch, long baseOffset, long logAppendTime, RuntimeException exception) {
    if (transactionManager != null) {
        if (exception instanceof OutOfOrderSequenceException
                && !transactionManager.isTransactional()
                && transactionManager.hasProducerId(batch.producerId())) {
            log.error("The broker received an out of order sequence number for topic-partition " +
                            "{} at offset {}. This indicates data loss on the broker, and should be investigated.",
                    batch.topicPartition, baseOffset);

            // Reset the transaction state since we have hit an irrecoverable exception and cannot make any guarantees
            // about the previously committed message. Note that this will discard the producer id and sequence
            // numbers for all existing partitions.
            transactionManager.resetProducerId();
        } else if (exception instanceof ClusterAuthorizationException
                || exception instanceof TransactionalIdAuthorizationException
                || exception instanceof ProducerFencedException) {
            transactionManager.transitionToFatalError(exception);
        } else if (transactionManager.isTransactional()) {
            transactionManager.transitionToAbortableError(exception);
        }
    }
    batch.done(baseOffset, logAppendTime, exception);
    this.accumulator.deallocate(batch);
}
 

import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
@Override
public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
                                     String consumerGroupId) throws ProducerFencedException {
    verifyProducerState();
    verifyTransactionsInitialized();
    verifyNoTransactionInFlight();
    Objects.requireNonNull(consumerGroupId);
    if (offsets.size() == 0) {
        return;
    }
    Map<TopicPartition, OffsetAndMetadata> uncommittedOffsets = this.uncommittedConsumerGroupOffsets.get(consumerGroupId);
    if (uncommittedOffsets == null) {
        uncommittedOffsets = new HashMap<>();
        this.uncommittedConsumerGroupOffsets.put(consumerGroupId, uncommittedOffsets);
    }
    uncommittedOffsets.putAll(offsets);
    this.sentOffsets = true;
}
 

import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
@Override
public void commitTransaction() throws ProducerFencedException {
    verifyProducerState();
    verifyTransactionsInitialized();
    verifyNoTransactionInFlight();

    flush();

    this.sent.addAll(this.uncommittedSends);
    if (!this.uncommittedConsumerGroupOffsets.isEmpty())
        this.consumerGroupOffsets.add(this.uncommittedConsumerGroupOffsets);

    this.uncommittedSends.clear();
    this.uncommittedConsumerGroupOffsets = new HashMap<>();
    this.transactionCommitted = true;
    this.transactionAborted = false;
    this.transactionInFlight = false;

    ++this.commitCount;
}
 

import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
@Override
protected void recoverAndCommit(KafkaTransactionState transaction) {
	switch (semantic) {
		case EXACTLY_ONCE:
			try (FlinkKafkaProducer<byte[], byte[]> producer =
					initTransactionalProducer(transaction.transactionalId, false)) {
				producer.resumeTransaction(transaction.producerId, transaction.epoch);
				producer.commitTransaction();
			}
			catch (InvalidTxnStateException | ProducerFencedException ex) {
				// That means we have committed this transaction before.
				LOG.warn("Encountered error {} while recovering transaction {}. " +
					"Presumably this transaction has been already committed before",
					ex,
					transaction);
			}
			break;
		case AT_LEAST_ONCE:
		case NONE:
			break;
		default:
			throw new UnsupportedOperationException("Not implemented semantic");
	}
}
 

import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
private void checkForException() {
    if (sendException != null) {
        if (sendException instanceof ProducerFencedException) {
            throw (ProducerFencedException) sendException;
        }
        throw new StreamsException(String.format("%s exception caught when producing", logPrefix), sendException);
    }
}
 

import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
@Test
public void shouldCloseTaskAsZombieAndRemoveFromActiveTasksIfProducerGotFencedAtBeginTransactionWhenTaskIsResumed() throws Exception {
    builder.addSource("name", "topic").addSink("out", "output");

    final MockClientSupplier clientSupplier = new MockClientSupplier(applicationId);
    final StreamThread thread = new StreamThread(
        builder,
        new StreamsConfig(configProps(true)),
        clientSupplier,
        applicationId,
        clientId,
        processId,
        new Metrics(),
        new MockTime(),
        new StreamsMetadataState(builder, StreamsMetadataState.UNKNOWN_HOST),
        0);

    final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
    activeTasks.put(task1, task0Assignment);

    thread.setPartitionAssignor(new MockStreamsPartitionAssignor(activeTasks));

    thread.rebalanceListener.onPartitionsRevoked(null);
    thread.rebalanceListener.onPartitionsAssigned(task0Assignment);
    assertThat(thread.tasks().size(), equalTo(1));

    thread.rebalanceListener.onPartitionsRevoked(null);
    clientSupplier.producers.get(0).fenceProducer();
    try {
        thread.rebalanceListener.onPartitionsAssigned(task0Assignment);
        fail("should have thrown " + ProducerFencedException.class.getSimpleName());
    } catch (final ProducerFencedException e) { }

    assertTrue(thread.tasks().isEmpty());
}
 

import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
@Override
public void beginTransaction() throws ProducerFencedException {
    verifyProducerState();
    verifyTransactionsInitialized();
    this.transactionInFlight = true;
    this.transactionCommitted = false;
    this.transactionAborted = false;
    this.sentOffsets = false;
}
 

import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
@Override
public void abortTransaction() throws ProducerFencedException {
    verifyProducerState();
    verifyTransactionsInitialized();
    verifyNoTransactionInFlight();
    flush();
    this.uncommittedSends.clear();
    this.uncommittedConsumerGroupOffsets.clear();
    this.transactionCommitted = false;
    this.transactionAborted = true;
    this.transactionInFlight = false;
}
 

import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
private void verifyProducerState() {
    if (this.closed) {
        throw new IllegalStateException("MockProducer is already closed.");
    }
    if (this.producerFenced) {
        throw new ProducerFencedException("MockProducer is fenced.");
    }
}
 

import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
/**
 * Should be called before the start of each new transaction.
 *
 * @throws ProducerFencedException if another producer is with the same
 *                                 transactional.id is active.
 */
public void beginTransaction() throws ProducerFencedException {
    // Set the transactional bit in the producer.
    if (transactionManager == null)
        throw new IllegalStateException("Cannot use transactional methods without enabling transactions");
    transactionManager.beginTransaction();
}
 

import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
@Test
public void shouldThrowOnBeginTransactionsIfProducerGotFenced() {
    producer.initTransactions();
    producer.fenceProducer();
    try {
        producer.beginTransaction();
        fail("Should have thrown as producer is fenced off");
    } catch (ProducerFencedException e) { }
}
 

import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
@Test
public void shouldThrowOnSendIfProducerGotFenced() {
    producer.initTransactions();
    producer.fenceProducer();
    try {
        producer.send(null);
        fail("Should have thrown as producer is fenced off");
    } catch (ProducerFencedException e) { }
}
 

import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
@Test
public void shouldThrowOnFlushIfProducerGotFenced() {
    producer.initTransactions();
    producer.fenceProducer();
    try {
        producer.flush();
        fail("Should have thrown as producer is fenced off");
    } catch (ProducerFencedException e) { }
}
 

import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
@Test
public void shouldThrowOnSendOffsetsToTransactionIfProducerGotFenced() {
    producer.initTransactions();
    producer.fenceProducer();
    try {
        producer.sendOffsetsToTransaction(null, null);
        fail("Should have thrown as producer is fenced off");
    } catch (ProducerFencedException e) { }
}
 

import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
@Test
public void shouldThrowOnCommitTransactionIfProducerGotFenced() {
    producer.initTransactions();
    producer.fenceProducer();
    try {
        producer.commitTransaction();
        fail("Should have thrown as producer is fenced off");
    } catch (ProducerFencedException e) { }
}
 

import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
@Test
public void shouldThrowOnAbortTransactionIfProducerGotFenced() {
    producer.initTransactions();
    producer.fenceProducer();
    try {
        producer.abortTransaction();
        fail("Should have thrown as producer is fenced off");
    } catch (ProducerFencedException e) { }
}
 

import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
private void closeIgnoringProducerFenced(AutoCloseable autoCloseable) throws Exception {
	try {
		autoCloseable.close();
	}
	catch (Exception ex) {
		if (!(ex.getCause() instanceof ProducerFencedException)) {
			throw ex;
		}
	}
}
 

import org.apache.kafka.common.errors.ProducerFencedException; //导入依赖的package包/类
@Override
public void beginTransaction() throws ProducerFencedException {
  producer.beginTransaction();
}