title: Extend Kafka Connect SinkTask and start consuming from given offsets
body: <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>