explicitly flush all buffers when closing the kafka producer. fix stateful task test.