title: Using Kafka Connect HOWTO "commit offsets" as soon as a "put" is completed in SinkTask
body: <p>I am using Kafka Connect to get messages from a Kafka Broker (v0.10.2) and then sync it to a downstream service.</p>
  
  <p>Currently, I have code in <code>SinkTask#put</code> that will process the <code>SinkRecord</code> &amp; then persist it to the downstream service.</p>
  
  <p>A couple of key requirements,</p>
  
  <ol>
  <li>We need to make sure the messages are persisted to the downstream service AT LEAST once. </li>
  <li>If the downstream service throws an error or says it didn't process the message then we need to make sure that the messages are re-read again.</li>
  </ol>
  
  <p>So we thought we can rely on <code>SinkTask#flush</code> to effectively back out of committing offsets for that particular poll/cycle of received messages by throwing an exception or something that will tell Connect not to commit the offsets, but retry in the next poll.</p>
  
  <p>But as we found out <code>flush</code> is actually time-based &amp; is more or less independent of the polls &amp; it will commit the offsets when it reaches a certain time threshold.</p>
  
  <p>In 0.10.2 <code>SinkTask#preCommit</code> was introduced, so we thought we can use it for our purposes. But nowhere in the documentation it is mentioned that there is a 1:1 relationship between <code>SinkTask#put</code> &amp; <code>SinkTask#preCommit</code>.</p>
  
  <p>Since essentially we want to <code>commit offsets</code> as soon as a single <code>put succeeds</code>. And similarly, <strong>not</strong> commit the offsets, if that particular <code>put</code> failed.</p>
  
  <p>How to accomplish this, if not via <code>SinkTask#preCommit</code>?</p>