fixed a bug in the transactional mode that was causing only the first tuple in the batch to be sent to kafka, the rest were just getting thrown away. Oops.