title: Kafka - Delayed Queue implementation using high level consumer
body: <p>Want to implement a delayed consumer using the high level consumer api </p>
  
  <p>main idea:</p>
  
  <ul>
  <li>produce messages by key (each msg contains creation timestamp) this makes sure that each partition has ordered messages by produced time.</li>
  <li>auto.commit.enable=false (will explicitly commit after each message process)</li>
  <li>consume a message</li>
  <li>check message timestamp and check if enough time has passed</li>
  <li>process message (this operation will never fail)</li>
  <li><p>commit 1 offset</p>
  
  <pre><code>while (it.hasNext()) {
    val msg = it.next().message()
    //checks timestamp in msg to see delay period exceeded
    while (!delayedPeriodPassed(msg)) { 
       waitSomeTime() //Thread.sleep or something....
    }
    //certain that the msg was delayed and can now be handled
    Try { process(msg) } //the msg process will never fail the consumer
    consumer.commitOffsets //commit each msg
  }
  </code></pre></li>
  </ul>
  
  <p>some concerns about this implementation:</p>
  
  <ol>
  <li>commit each offset might slow ZK down</li>
  <li>can consumer.commitOffsets throw an exception? if yes i will consume the same message twice (can solve with idempotent messages)</li>
  <li>problem waiting long time without committing the offset, for example delay period is 24 hours, will get next from iterator, sleep for 24 hours, process and commit (ZK session timeout ?)</li>
  <li>how can ZK session keep-alive without commit new offsets ? (setting a hive zookeeper.session.timeout.ms can resolve in dead consumer without recognising it)</li>
  <li>any other problems im missing?</li>
  </ol>
  
  <p>Thanks!</p>