flume-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From <j.guilm...@accenture.com>
Subject RE: runner.backoffs and runner.backoffs.consecutive ? related to ChannelFullException ?
Date Tue, 13 Jan 2015 18:55:05 GMT
Hi,

As far as I can understand, this mechanism is involved in the SinkRunner, which is basically
a Thread in charge of processing the event to be sent by the sinks.

It is an infinite loop, with a backoff mechanism in order to avoid trying too long to use
a deficient sink.

If backoff is enabled then the client will temporarily blacklist hosts that fail, causing
them to be excluded from being selected as a failover host until a given timeout. When the
timeout elapses, if the host is still unresponsive then this is considered a sequential failure,
and the timeout is increased exponentially to avoid potentially getting stuck in long waits
on unresponsive hosts.

The maximum backoff time can be configured by setting maxBackoff (in milliseconds). The maxBackoff
default is 30 seconds (specified in the OrderSelector class that's the superclass of both
load balancing strategies). The backoff timeout will increase exponentially with each sequential
failure up to the maximum possible backoff timeout. The maximum possible backoff is limited
to 65536 seconds (about 18.2 hours).
A focus on the code explains a bit more:

    public void run() {
      logger.debug("Polling sink runner starting");

      while (!shouldStop.get()) {
        try {
          if (policy.process().equals(Sink.Status.BACKOFF)) {
            counterGroup.incrementAndGet("runner.backoffs");

            Thread.sleep(Math.min(
                counterGroup.incrementAndGet("runner.backoffs.consecutive")
                * backoffSleepIncrement, maxBackoffSleep));
          } else {
            counterGroup.set("runner.backoffs.consecutive", 0L);
          }
        } catch (InterruptedException e) {
          logger.debug("Interrupted while processing an event. Exiting.");
          counterGroup.incrementAndGet("runner.interruptions");
        } catch (Exception e) {
          logger.error("Unable to deliver event. Exception follows.", e);
          if (e instanceof EventDeliveryException) {
            counterGroup.incrementAndGet("runner.deliveryErrors");
          } else {
            counterGroup.incrementAndGet("runner.errors");
          }
          try {
            Thread.sleep(maxBackoffSleep);
          } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
          }
        }
      }
      logger.debug("Polling runner exiting. Metrics:{}", counterGroup);
    }

  }


ð  Apparently your sink is existing with BACKOFF status, and therefore Flume wait more and
more (until max) before retrying, hoping the sink will recover.

Jeff

From: terreyshih [mailto:terreyshih@gmail.com]
Sent: mardi 13 janvier 2015 19:20
To: user@flume.apache.org
Subject: runner.backoffs and runner.backoffs.consecutive ? related to ChannelFullException
?

anybody and shed some light on this ? thx

On Jan 12, 2015, at 11:28 PM, terreyshih <terreyshih@gmail.com<mailto:terreyshih@gmail.com>>
wrote:

Hi,

What do the values runner.backoffs.consecutive and runner.backoffs value mean ? I get ChannelFullException
shortly afterwards.

thanks,
-Terrey



2015-01-10 02:05:14,500 DEBUG [lifecycleSupervisor-1-1] [o.a.f.l.LifecycleSupervisor.run:214]
- checking process:SinkRunner: { policy:org.apache.flume.sink.DefaultSinkProcessor@63b8dba5
counterGroup:{ name:null counters:{runner.backoffs.consecutive=8, runner.backoffs=2120} }
} supervisoree:{ status:{ lastSeen:1420884311499 lastSeenState:START desiredState:START firstSeen:1420869500254
failures:0 discard:false error:false } policy:org.apache.flume.lifecycle.LifecycleSupervisor$SupervisorPolicy$AlwaysRestartPolicy@2e60639b
}



Caused by: org.apache.flume.ChannelFullException: Space for commit to queue couldn't be acquired.
Sinks are likely not keeping up with sources, or the buffer size is too tight


________________________________

This message is for the designated recipient only and may contain privileged, proprietary,
or otherwise confidential information. If you have received it in error, please notify the
sender immediately and delete the original. Any other use of the e-mail by you is prohibited.
Where allowed by local law, electronic communications with Accenture and its affiliates, including
e-mail and instant messaging (including content), may be scanned by our systems for the purposes
of information security and assessment of internal compliance with Accenture policy.
______________________________________________________________________________________

www.accenture.com

Mime
View raw message