foo

is Iterable<*>

T

org.apache.avro.specific.SpecificRecord

private var stateProducer = KafkaProducer<String, State>(Settings.getProducer("api-states"))

and then

stateProducer.send(ProducerRecord("states", machineId, stateRecord))

private fun producerFor(thing: SpecificRecord): ProducerType { 
    val klass = thing::class
    when (klass) {
        is Metric -> metricProducer,
        is Event -> eventProducer,
        ...
}

...each(
  producer = producerFot(it)
  producer.startTransaction()
  producer.send(it)
  producer.commitTransaction()
)

producer.startTransaction()
list.each(
  producer.send(it)
)
producer.commitTransaction()

yes but the type within a single iterator is always the same

producer.transaction {
    list.each {
        producer.send(this)
    }
}

class Producer {
    fun transaction(lambda: () -> Boolean) {
        startTransaction()
        lambda()
        commitTransaction()
    }
}

fun Producer.transaction(...) {

private fun <K, V> wrapInTransaction(producer: KafkaProducer<K, V>, lambda: (producer: KafkaProducer<K, V>) -> Unit) {
    try {
        producer.beginTransaction()
        lambda(producer)
        producer.commitTransaction()
    } catch (e: ProducerFencedException) {
        eventProducer.close()
    } catch (e: OutOfOrderSequenceException) {
        eventProducer.close()
    } catch (e: AuthorizationException) {
        eventProducer.close()
    } catch (e: KafkaException) {
       eventProducer.abortTransaction()
    }
}

foo.x()
foo.y()

apply

empty

empty

apply

foo.x()
foo.y()
// becomes
foo.apply {
  x()
  y()
}

with(foo) {
  x()
  y()
}

with