Fix issue with restarts rereading kafka messages

Change-Id: I54265a4fcfb70dcf53f8d7db14290e89f009aec1