title: Kafka KStreams - processing timeouts
body: <p>I am attempting to use <code>&lt;KStream&gt;.process()</code> with a <code>TimeWindows.of(&quot;name&quot;, 30000)</code> to batch up some <strong>KTable</strong> values and send them on. It seems that 30 seconds exceeds the consumer timeout interval after which Kafka considers said consumer to be defunct and releases the partition.</p>
  <p>I've tried upping the frequency of <strong>poll</strong> and <strong>commit interval</strong> to avoid this:</p>
  <pre><code>config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, &quot;5000&quot;);
  config.put(StreamsConfig.POLL_MS_CONFIG, &quot;5000&quot;);
  </code></pre>
  <p>Unfortunately these errors are still occurring:</p>
  <p>(lots of these)</p>
  <pre><code>ERROR  o.a.k.s.p.internals.RecordCollector - Error sending record to topic kafka_test1-write_aggregate2-changelog 
  org.apache.kafka.common.errors.TimeoutException: Batch containing 1 record(s) expired due to timeout while requesting metadata from brokers for kafka_test1-write_aggregate2-changelog-0
  </code></pre>
  <p>Followed by these:</p>
  <pre><code>INFO   o.a.k.c.c.i.AbstractCoordinator - Marking the coordinator 12.34.56.7:9092 (id: 2147483547 rack: null) dead for group kafka_test1
  WARN   o.a.k.s.p.internals.StreamThread - Failed to commit StreamTask #0_0 in thread [StreamThread-1]: 
    org.apache.kafka.clients.consumer.CommitFailedException: 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 session.timeout.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.
  at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:578)
  </code></pre>
  <p>Clearly I need to be sending heartbeats back to the server more often. How?</p>
  <p>My topology is:</p>
  <pre><code>KStreamBuilder kStreamBuilder = new KStreamBuilder();
  KStream&lt;String, String&gt; lines = kStreamBuilder.stream(TOPIC);
  KTable&lt;Windowed&lt;String&gt;, String&gt;  kt = lines.aggregateByKey(
              new DBAggregateInit(),
              new DBAggregate(),
              TimeWindows.of(&quot;write_aggregate2&quot;, 30000));
  
  DBProcessorSupplier dbProcessorSupplier = new DBProcessorSupplier();
  
  kt.toStream().process(dbProcessorSupplier);
  KafkaStreams kafkaStreams = new KafkaStreams(kStreamBuilder, streamsConfig);
  
  kafkaStreams.start();
  </code></pre>
  <p>The <strong>KTable</strong> is grouping values by key every 30 seconds. In <strong>Processor.init()</strong> I call <code>context.schedule(30000)</code>.</p>
  <p><strong>DBProcessorSupplier</strong> provides an instance of <strong>DBProcessor</strong>. This is an implementation of <strong>AbstractProcessor</strong> where all the overrides have been provided. All they do is LOG so I know when each is being hit.</p>
  <p>It's a pretty simple topology but it's clear I'm missing a step somewhere.</p>
  <hr />
  <h2>Edit:</h2>
  <p>I get that I can adjust this on the server side but Im hoping there is a client-side solution. I like the notion of partitions being made available pretty quickly when a client exits / dies.</p>
  <hr />
  <h2>Edit:</h2>
  <p>In an attempt to simplify the problem I removed the aggregation step from the graph. It's now just consumer-&gt;processor(). (If I send the consumer directly to <code>.print()</code> it works v quickly so I know it's ok). (Similarly If I output the aggregation (KTable) via <code>.print()</code> it seems ok too).</p>
  <p>What I found was that the <code>.process()</code> - which <em>should</em> be calling <code>.punctuate()</code> every 30 seconds is actually blocking for variable lengths of time and outputting somewhat randomly (if at all).</p>
  <ul>
  <li><a href="https://gist.github.com/ethrbunny/ab24056e25667a5a9e082bdf580a21ad" rel="noreferrer">Main program</a></li>
  <li><a href="https://gist.github.com/ethrbunny/387a5a75301694623adea052646281c0" rel="noreferrer">Debug output</a></li>
  <li><a href="https://gist.github.com/ethrbunny/9ebb4e3e2b9babdea124fe8a6a7cf5cf" rel="noreferrer">Processor Supplier</a></li>
  <li><a href="https://gist.github.com/ethrbunny/8355e62e043b8125136fab4668312730" rel="noreferrer">Processor</a></li>
  </ul>
  <h2>Further:</h2>
  <p>I set the debug level to 'debug' and reran. Im seeing lots of messages:</p>
  <pre><code>DEBUG  o.a.k.s.p.internals.StreamTask - Start processing one record [ConsumerRecord &lt;info&gt;
  </code></pre>
  <p>but a breakpoint in the <code>.punctuate()</code> function isn't getting hit. So it's doing lots of work but not giving me a chance to use it.</p>