es_sink: pass key=>(offset, value) instead of just key=>value as event metadata from stream tags