BackOffPolicy does not work because error count is resetted

[dwirtz]

The AmqpConsumer uses a backoff policy when there is any channel error or problems like backpressure until the channel reaches its capacity and gets full. In this case the consumer restarts and wait up to backoffTime before consuming data via amqp again. The backoffTime is calculated based on the counter numberOfConsecutiveConnectionErrors. Unfortunateley in this case (channel gets full) this counter is always resetted for each retry. In consequence the counter is always equals to 1 and therefore the backOffTime is always 1000 ms.

See consumeLoop method which resets the counter each time:

flume-ng-amqp-plugin/src/main/java/org/apache/flume/amqp/AmqpConsumer.java

Lines 165 to 211 in b253ba6

	private void consumeLoop() throws InterruptedException {
	Thread currentThread = Thread.currentThread();
	Channel channel = null;
	Connection connection = null;
	int numberOfConsecutiveConnectionErrors = 0;
	while (!currentThread.isInterrupted()) {
	try {
	connection = createConnection();
	// successfully connected, reset counter
	numberOfConsecutiveConnectionErrors = 0;
	channel = createChannel(connection);
	declarationsForChannel(channel);
	QueueingConsumer consumer = new QueueingConsumer(channel);
	processDeliveriesForConsumer(currentThread, consumer);
	} catch (InterruptedException e) {
	// NOTE - we need to catch InterruptedException before Exception to ensure we see the shutdown signal
	LOG.info("Received interrupt, shutting down consumer");
	// NEED to set the interrupt status on the thread to ensure the loop terminates
	currentThread.interrupt();
	} catch (IOException e) {
	LOG.info("IOException caught. Closing connection and waiting to reconnect", e);
	numberOfConsecutiveConnectionErrors++;
	} catch (Exception e) {
	// NOTE - this should never happen, but we don't want the thread to die because of an uncaught exception
	LOG.error("Consumer encountered an exception while processing the events", e);
	}
	// we need to take the interrupt status BEFORE closing because the close methods can alter the INTERRUPT state of the thread
	boolean wasInterrupted = currentThread.isInterrupted();
	closeChannelSilently(channel);
	closeConnectionSilently(connection);
	if (wasInterrupted) {
	currentThread.interrupt();
	}
	if (!currentThread.isInterrupted()) {
	waitToRetry(Sleeper.THREAD_SLEEPER, numberOfConsecutiveConnectionErrors);
	}
	}
	}

Either the counter must not be resetted within the while loop or the back off time calculation should be independent from that counter.