updated kafka writer