Add transaction / bulk send support in producer

[er1c]

Some psuedo code, that doesn't actually work because initTransactions() needs to only be called once

  final def produceWithTransaction[T <: Topic, M: Encoder: SchemaFor](
    topic: T,
    partitionKey: M => String,
    messages: Iterable[M]
  )(implicit topicSchema: TopicSchema[T, M]): Future[Iterable[ProducerResult[T, M]]] = {
    // This is the workflow documented in http://kafka.apache.org/0110/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html
    initTransactions()

    try {
      beginTransaction()
      // TODO: messages.par.map...???
      Future.sequence(messages.map { m => produce(topic, partitionKey(m), m) }).map{ ret =>
        commitTransaction()
        ret
      }
    } catch {
      case ex if Producer.isFatalKafkaException(ex) =>
        // We can't recover from these exceptions, so our only option is to close the producer and exit.
        close()
        Future.failed(ex)
      case ex: KafkaException =>
        abortTransaction()
        Future.failed(ex)
    }
  }

  private def isFatalKafkaException(ex: Throwable): Boolean = ex match {
    case _: ProducerFencedException => true
    case _: OutOfOrderSequenceException => true
    case _: AuthorizationException => true
    case _ => false
  }