title: spring-kafka processing the same message multiple times with very long tasks.
body: <p>This has been asked and addressed a couple of time but with my very limited knowledge I'm unable to find the answer to my problem. </p>
  
  <p>I have a producer which send a work task to a consumer, the task takes about two hours to complete. I need the task to just execute once, however it finishes then starts again over and over. </p>
  
  <p>The most helpful thing I've found my logs is </p>
  
  <pre><code>2018-05-15 15:18:23.731  WARN 6888 --- [container-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-2, groupId=anonymous.1ae85859-db41-4dc2-a7e2-ab4268256e00] Synchronous auto-commit of offsets {consumer-message-0=OffsetAndMetadata{offset=34, metadata=''}} failed: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records.
  </code></pre>
  
  <p>Which lead me to think that simply wrapping the consumer task in a thread would solve, however it hasn't. </p>
  
  <p>Some code from my consumer</p>
  
  <pre><code>@Component
  @Slf4j
  public class KafkaConsumer {
  
      private final CommandRunnerService commandRunnerService;
  
      public KafkaConsumer(CommandRunnerService commandRunnerService) {
          this.commandRunnerService = commandRunnerService;
      }
  
      @StreamListener(KafkaStreams.INPUT)
      public void handleWorkUnit(@Payload Steak steak) {
          commandRunnerService.executeCreateSteak(steak);
      }
  }
  </code></pre>
  
  <p>It's the <code>handleWorkUnit</code> which takes a few hours to complete. So my attempt to fix was </p>
  
  <pre><code>    @StreamListener(KafkaStreams.INPUT)
      public void handleWorkUnit(@Payload Steak steak) {
          Runnable task = () -&gt; commandRunnerService.executeCreateSteak(steak);
          task.run();
      }
  </code></pre>
  
  <p>Which made no difference. </p>
  
  <p>I'm using out-the-box configurations with only the very basics set on the consumer </p>
  
  <pre><code>spring:
    application:
    cloud:
      stream:
        kafka:
          binder:
            brokers: 192.168.0.100
        bindings:
          consumer-message:
            destination: consumer-message
            contentType: application/json
          consumer-response:
            destination: consumer-response
            contentType: application/json
  </code></pre>
  
  <p>And the versions of things I'm using:</p>
  
  <pre><code>ext {
      springCloudVersion = 'Finchley.RC1'
  }
  
  dependencies {
      compile('org.springframework.cloud:spring-cloud-stream')
      compile('org.springframework.cloud:spring-cloud-stream-binder-kafka')
      compile('org.springframework.kafka:spring-kafka')
  }
  </code></pre>
  
  <p>As mentioned above I've seen many complicated examples in both the docs and on SO, but I'm hoping for a simple config fix? Or some more "beginner" friendly examples. </p>
  
  <p>Cheers, </p>