Major Description
The call to commit offsets appears ok at first glance because we've invoked the SourceTask's stop method, but it runs before we've stopped the work thread which could still be invoking the SourceTask's poll() method and may have outstanding data. We need to wait until we're sure the work thread has completely finished so we're guaranteed to flush all the data generated by the SourceTask and commit the final offsets.