@Override
  public void stop() {
    if (!stopped.compareAndSet(false, true)) {
      LOG.warn("{}: Attempting to stop stopped consumer.", this);
      return;
    }

    LOG.info("{}: Stopping Samza kafkaConsumer ", this);

    // stop the proxy (with 1 minute timeout)
    if (proxy != null) {
      LOG.info("{}: Stopping proxy {}", this, proxy);
      proxy.stop(TimeUnit.SECONDS.toMillis(60));
    }

    try {
      synchronized (kafkaConsumer) {
        LOG.info("{}: Closing kafkaSystemConsumer {}", this, kafkaConsumer);
        kafkaConsumer.close();
      }
    } catch (Exception e) {
      LOG.warn("{}: Failed to stop KafkaSystemConsumer.", this, e);
    }
  }
Invoked by {@link KafkaConsumerProxy} to notify the consumer of failure, so it can relay and stop the BEM polling.
@param throwable the cause of the failure of the proxy
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemConsumer.java