title: How to close Flume with kafka as Source
body: <p>I'm upgrading Flume1.7 to 1.9.I have 5 Kafka sources and 7 aws-s3's sinks in FLume's conf.
    Firstly I need stop Flume 1.7.So I execute the command
  'kill <code>ps -ef |grep flume |grep bidinfo | awk '{print $2}'</code>' to stop the bidinfo task,but this process still exists.22 hours later ,this process always alive until now. What else can I do without this command 'kill -9 xxxx'.Welcome to suggest!!!</p>
  
  <p>This server has been running for 60 days with Flume 1.7,kafka ,I've tried to execute the command 'kill -3 xxxx',However, only two sources and two sinks have been stopped and the others have been running. 
    I read the source code of flume-kafkaSource and observed flume's log.
  It is possible that two methods(Consumer. wakeup ();Consumer. close ();) are not completed.</p>
  
  <pre><code>#flume-conf
  ag.sources = src_1 src_2 src_3 src_4 src_5
  ag.channels = ch
  ag.sinks = sk_1 sk_2 sk_3 sk_4 sk_5 sk_6 sk_7
  
  #source:kafka
  ag.sources.src_1.type = org.apache.flume.source.kafka.KafkaSource
  ag.sources.src_1.kafka.bootstrap.servers = xxxx
  ag.sources.src_1.kafka.consumer.group.id = flume.xxxxx
  ag.sources.src_1.kafka.consumer.retry.backoff.ms = 10000
  ag.sources.src_1.batchSize = 5000
  ag.sources.src_1.batchDurationMillis = 2000
  ag.sources.src_1.kafka.topics = xxxx
  ag.sources.src_1.interceptors = i1
  ag.sources.src_1.interceptors.i1.type = xxxx.interceptor
  ag.sources.src_1.channels = ch
  
  #sink:aws S3
  ag.sinks.sk_1.type = hdfs
  ag.sinks.sk_1.hdfs.path = s3a://xxxxxxx
  ag.sinks.sk_1.hdfs.filePrefix = %{minute}
  ag.sinks.sk_1.hdfs.fileSuffix = .xxx.1.lzo
  ag.sinks.sk_1.hdfs.rollSize = 0
  ag.sinks.sk_1.hdfs.rollCount = 0
  ag.sinks.sk_1.hdfs.rollInterval = 0
  ag.sinks.sk_1.hdfs.idleTimeout = 180
  ag.sinks.sk_1.hdfs.callTimeout = 600000
  ag.sinks.sk_1.hdfs.closeTries = 5
  ag.sinks.sk_1.hdfs.retryInterval = 60
  ag.sinks.sk_1.hdfs.batchSize = 3000
  ag.sinks.sk_1.hdfs.codeC = lzop
  ag.sinks.sk_1.hdfs.fileType = CompressedStream
  ag.sinks.sk_1.hdfs.writeFormat = Text
  ag.sinks.sk_1.channel = ch
  #channels
  
  ag.channels.ch.type = memory
  ag.channels.ch.capacity = 2000000
  ag.channels.ch.transactionCapacity = 100000
  </code></pre>
  
  <h1>Error Message</h1>
  
  <pre><code>09 Sep 2019 11:14:18,010 ERROR [PollableSourceRunner-KafkaSource-src_2] (org.apache.flume.source.kafka.KafkaSource.doProcess:314)  - KafkaSource EXCEPTION, {}
  org.apache.flume.ChannelException: java.lang.InterruptedException
          at org.apache.flume.channel.BasicTransactionSemantics.commit(BasicTransactionSemantics.java:154)
          at org.apache.flume.channel.ChannelProcessor.processEventBatch(ChannelProcessor.java:194)
          at org.apache.flume.source.kafka.KafkaSource.doProcess(KafkaSource.java:295)
          at org.apache.flume.source.AbstractPollableSource.process(AbstractPollableSource.java:60)
          at org.apache.flume.source.PollableSourceRunner$PollingRunner.run(PollableSourceRunner.java:133)
          at java.lang.Thread.run(Thread.java:745)
  Caused by: java.lang.InterruptedException
          at java.util.concurrent.locks.AbstractQueuedSynchronizer.tryAcquireSharedNanos(AbstractQueuedSynchronizer.java:1326)
          at java.util.concurrent.Semaphore.tryAcquire(Semaphore.java:582)
          at org.apache.flume.channel.MemoryChannel$MemoryTransaction.doCommit(MemoryChannel.java:119)
          at org.apache.flume.channel.BasicTransactionSemantics.commit(BasicTransactionSemantics.java:151)
          ... 5 more
  #this place I think is the main problem
  
  09 Sep 2019 11:14:18,011 INFO  [PollableSourceRunner-KafkaSource-src_2] (org.apache.flume.source.PollableSourceRunner$PollingRunner.run:143)  - Source runner interrupted. Exiting
  09 Sep 2019 11:14:18,011 INFO  [agent-shutdown-hook] (org.apache.kafka.clients.consumer.internals.AbstractCoordinator$2.onFailure:571)  - LeaveGroup request failed with error
  org.apache.kafka.clients.consumer.internals.SendFailedException
  
  #-------------
  09 Sep 2019 11:14:18,018 INFO  [agent-shutdown-hook] (org.apache.flume.instrumentation.MonitoredCounterGroup.stop:149)  - Component type: SOURCE, name: src_2 stopped
  09 Sep 2019 11:14:18,018 INFO  [agent-shutdown-hook] (org.apache.flume.instrumentation.MonitoredCounterGroup.stop:155)  - Shutdown Metric for type: SOURCE, name: src_2. source.start.time == 1567911973591
  09 Sep 2019 11:14:18,018 INFO  [agent-shutdown-hook] (org.apache.flume.instrumentation.MonitoredCounterGroup.stop:161)  - Shutdown Metric for type: SOURCE, name: src_2. source.stop.time == 1567998858018
  09 Sep 2019 11:14:18,018 INFO  [agent-shutdown-hook] (org.apache.flume.instrumentation.MonitoredCounterGroup.stop:177)  - Shutdown Metric for type: SOURCE, name: src_2. source.kafka.commit.time == 2470886
  09 Sep 2019 11:14:18,018 INFO  [agent-shutdown-hook] (org.apache.flume.instrumentation.MonitoredCounterGroup.stop:177)  - Shutdown Metric for type: SOURCE, name: src_2. source.kafka.empty.count == 0
  09 Sep 2019 11:14:18,018 INFO  [agent-shutdown-hook] (org.apache.flume.instrumentation.MonitoredCounterGroup.stop:177)  - Shutdown Metric for type: SOURCE, name: src_2. source.kafka.event.get.time == 78823378
  09 Sep 2019 11:14:18,018 INFO  [agent-shutdown-hook] (org.apache.flume.instrumentation.MonitoredCounterGroup.stop:177)  - Shutdown Metric for type: SOURCE, name: src_2. src.append-batch.accepted == 0
  09 Sep 2019 11:14:18,018 INFO  [agent-shutdown-hook] (org.apache.flume.instrumentation.MonitoredCounterGroup.stop:177)  - Shutdown Metric for type: SOURCE, name: src_2. src.append-batch.received == 0
  09 Sep 2019 11:14:18,018 INFO  [agent-shutdown-hook] (org.apache.flume.instrumentation.MonitoredCounterGroup.stop:177)  - Shutdown Metric for type: SOURCE, name: src_2. src.append.accepted == 0
  09 Sep 2019 11:14:18,018 INFO  [agent-shutdown-hook] (org.apache.flume.instrumentation.MonitoredCounterGroup.stop:177)  - Shutdown Metric for type: SOURCE, name: src_2. src.append.received == 0
  09 Sep 2019 11:14:18,018 INFO  [agent-shutdown-hook] (org.apache.flume.instrumentation.MonitoredCounterGroup.stop:177)  - Shutdown Metric for type: SOURCE, name: src_2. src.events.accepted == 767319160
  09 Sep 2019 11:14:18,018 INFO  [agent-shutdown-hook] (org.apache.flume.instrumentation.MonitoredCounterGroup.stop:177)  - Shutdown Metric for type: SOURCE, name: src_2. src.events.received == 767366357
  09 Sep 2019 11:14:18,018 INFO  [agent-shutdown-hook] (org.apache.flume.instrumentation.MonitoredCounterGroup.stop:177)  - Shutdown Metric for type: SOURCE, name: src_2. src.open-connection.count == 0
  </code></pre>