Extend Kafka Connect SinkTask and start consuming from given offsets
<p>I want to extend <code>SinkTask</code> to create my own sink connector.</p>

<p>If I save the offsets during the flush, and the next time I start the sink connector I would like to resume reading from my saved offsets, what would be the correct way to do it?</p>

<p>I tried using the <code>SinkTaskContext</code> of the overridden <code>initialize(SinkTaskContext context)</code> to assign my own offsets:</p>

<pre class="lang-java prettyprint-override"><code>@Override
public void initialize(SinkTaskContext context) {
  HashMap&lt;TopicPartition, Long&gt; offsetMap = new HashMap&lt;&gt;();
  ...
  context.offset(offsetMap);
}
</code></pre>

<p>But this doesn’t work, because the partitions are not assigned yet. I was getting an exception.</p>

<p>Should I then save the context (from <code>initialize()</code>) into a global variable and then use it to assign the offsets to it inside the method <code>open(Collection&lt;TopicPartition&gt; partitions)</code> (overridden from <code>SinkTask</code>) in the same way I was doing inside <code>initialize</code>? e.g.:</p>

<pre class="lang-java prettyprint-override"><code>@Override
public void open(Collection&lt;TopicPartition&gt; partitions) {
  HashMap&lt;TopicPartition, Long&gt; offsetMapNew = new HashMap&lt;&gt;();
  for (TopicPartition tp : partitions) // for each partition assigned
  {
     Long offset = myOffsetMap.get(tp.topic() + "-" + tp.partition());
     if (offset == null) { offset = 0l; } // 0 Long
     offsetMapNew.put(tp, offset);
  }
  mySavedTaskContext.offset(offsetMapNew); // sync offsets ?
}
</code></pre>