diff --git a/README.md b/README.md
index c296e2f4..e1b76273 100644
--- a/README.md
+++ b/README.md
@@ -52,19 +52,18 @@ dependencies {
 
 <!--- INCLUDE
 import arrow.continuations.SuspendApp
-import io.github.nomisRev.kafka.Acks
 import io.github.nomisRev.kafka.Admin
 import io.github.nomisRev.kafka.AdminSettings
-import io.github.nomisRev.kafka.ProducerSettings
 import io.github.nomisRev.kafka.createTopic
 import io.github.nomisRev.kafka.imap
 import io.github.nomisRev.kafka.map
-import io.github.nomisRev.kafka.produce
+import io.github.nomisRev.kafka.publisher.Acks
+import io.github.nomisRev.kafka.publisher.KafkaPublisher
+import io.github.nomisRev.kafka.publisher.PublisherSettings
 import io.github.nomisRev.kafka.receiver.AutoOffsetReset
 import io.github.nomisRev.kafka.receiver.KafkaReceiver
 import io.github.nomisRev.kafka.receiver.ReceiverSettings
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.flow.asFlow
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.take
 import kotlinx.coroutines.launch
@@ -94,17 +93,19 @@ fun main(): Unit = SuspendApp {
   }
 
   launch(Dispatchers.IO) { // Send 20 messages, and then close the producer
-    val settings: ProducerSettings<Key, Message> = ProducerSettings(
+    val settings: PublisherSettings<Key, Message> = PublisherSettings(
       kafka.bootstrapServers,
       IntegerSerializer().imap { key: Key -> key.index },
       StringSerializer().imap { msg: Message -> msg.content },
       Acks.All
     )
-    (1..msgCount)
-      .asFlow()
-      .map { index -> ProducerRecord(topicName, Key(index), Message("msg: $index")) }
-      .produce(settings)
-      .collect(::println)
+    KafkaPublisher(settings).use { publisher ->
+      publisher.publishScope {
+        (1..msgCount).forEach { index ->
+          offer(ProducerRecord(topicName, Key(index), Message("msg: $index")))
+        }
+      }
+    }
   }
 
   launch(Dispatchers.IO) { // Consume 20 messages as a stream, and then close the consumer
@@ -127,16 +128,6 @@ fun main(): Unit = SuspendApp {
 > You can get the full code [here](guide/example/example-readme-01.kt).
 
 ```text
-test-topic-0@0
-test-topic-0@1
-test-topic-0@2
-test-topic-0@3
-test-topic-0@4
-test-topic-0@5
-test-topic-0@6
-test-topic-0@7
-test-topic-0@8
-test-topic-0@9
 Key(index=1) -> Message(content=msg: 1)
 Key(index=2) -> Message(content=msg: 2)
 Key(index=3) -> Message(content=msg: 3)
diff --git a/guide/example/example-readme-01.kt b/guide/example/example-readme-01.kt
index 32f66b6d..e90ed3e0 100644
--- a/guide/example/example-readme-01.kt
+++ b/guide/example/example-readme-01.kt
@@ -1,19 +1,18 @@
 package example.exampleReadme01
 
 import arrow.continuations.SuspendApp
-import io.github.nomisRev.kafka.Acks
 import io.github.nomisRev.kafka.Admin
 import io.github.nomisRev.kafka.AdminSettings
-import io.github.nomisRev.kafka.ProducerSettings
 import io.github.nomisRev.kafka.createTopic
 import io.github.nomisRev.kafka.imap
 import io.github.nomisRev.kafka.map
-import io.github.nomisRev.kafka.produce
+import io.github.nomisRev.kafka.publisher.Acks
+import io.github.nomisRev.kafka.publisher.KafkaPublisher
+import io.github.nomisRev.kafka.publisher.PublisherSettings
 import io.github.nomisRev.kafka.receiver.AutoOffsetReset
 import io.github.nomisRev.kafka.receiver.KafkaReceiver
 import io.github.nomisRev.kafka.receiver.ReceiverSettings
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.flow.asFlow
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.take
 import kotlinx.coroutines.launch
@@ -41,17 +40,19 @@ fun main(): Unit = SuspendApp {
   }
 
   launch(Dispatchers.IO) { // Send 20 messages, and then close the producer
-    val settings: ProducerSettings<Key, Message> = ProducerSettings(
+    val settings: PublisherSettings<Key, Message> = PublisherSettings(
       kafka.bootstrapServers,
       IntegerSerializer().imap { key: Key -> key.index },
       StringSerializer().imap { msg: Message -> msg.content },
       Acks.All
     )
-    (1..msgCount)
-      .asFlow()
-      .map { index -> ProducerRecord(topicName, Key(index), Message("msg: $index")) }
-      .produce(settings)
-      .collect(::println)
+    KafkaPublisher(settings).use { publisher ->
+      publisher.publishScope {
+        (1..msgCount).forEach { index ->
+          offer(ProducerRecord(topicName, Key(index), Message("msg: $index")))
+        }
+      }
+    }
   }
 
   launch(Dispatchers.IO) { // Consume 20 messages as a stream, and then close the consumer
diff --git a/guide/src/main/kotlin/main.kt b/guide/src/main/kotlin/main.kt
deleted file mode 100644
index be158097..00000000
--- a/guide/src/main/kotlin/main.kt
+++ /dev/null
@@ -1,73 +0,0 @@
-package io.github.nomisRev.kafka
-
-import io.github.nomisRev.kafka.receiver.KafkaReceiver
-import io.github.nomisRev.kafka.receiver.ReceiverSettings
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.coroutineScope
-import kotlinx.coroutines.delay
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.asFlow
-import kotlinx.coroutines.flow.map
-import kotlinx.coroutines.flow.take
-import kotlinx.coroutines.launch
-import kotlinx.coroutines.runBlocking
-import org.apache.kafka.clients.admin.NewTopic
-import org.apache.kafka.clients.producer.ProducerRecord
-import org.apache.kafka.common.serialization.IntegerDeserializer
-import org.apache.kafka.common.serialization.IntegerSerializer
-import org.apache.kafka.common.serialization.StringDeserializer
-import org.apache.kafka.common.serialization.StringSerializer
-import java.util.UUID
-
-@JvmInline
-value class Key(val index: Int)
-
-@JvmInline
-value class Message(val content: String)
-
-fun main(): Unit = runBlocking(Dispatchers.Default) {
-  val topicName = "test-topic"
-  val msgCount = 25
-  val kafka = Kafka.container
-  
-  Admin(AdminSettings(kafka.bootstrapServers)).use { client ->
-    client.createTopic(NewTopic(topicName, 1, 1))
-  }
-  
-  coroutineScope { // Run produces and consumer in a single scope
-    launch(Dispatchers.IO) { // Send 20 messages, and then close the producer
-      val settings: ProducerSettings<Key, Message> = ProducerSettings(
-        kafka.bootstrapServers,
-        IntegerSerializer().imap { key: Key -> key.index },
-        StringSerializer().imap { msg: Message -> msg.content },
-        Acks.All
-      )
-      (1..msgCount)
-        .asFlow()
-        .map { index ->
-          delay(index * 50L)
-          ProducerRecord(topicName, Key(index), Message("msg: $index"))
-        }
-        .produce(settings)
-        .collect(::println)
-    }
-    
-    launch(Dispatchers.IO) { // Consume 20 messages as a stream, and then close the consumer
-      val settings: ReceiverSettings<Key, Message> = ReceiverSettings(
-        kafka.bootstrapServers,
-        IntegerDeserializer().map(::Key),
-        StringDeserializer().map(::Message),
-        groupId = UUID.randomUUID().toString(),
-        autoOffsetReset = AutoOffsetReset.Earliest
-      )
-      KafkaReceiver(settings)
-        .receive(topicName)
-        .take(msgCount)
-        .collect {
-          delay(75)
-          println("${Thread.currentThread().name} => ${it.key()} -> ${it.value()}")
-          it.offset.acknowledge()
-        }
-    }
-  }
-}
diff --git a/src/main/kotlin/io/github/nomisRev/kafka/publisher/Acks.kt b/src/main/kotlin/io/github/nomisRev/kafka/publisher/Acks.kt
index 74e64a99..f28c545e 100644
--- a/src/main/kotlin/io/github/nomisRev/kafka/publisher/Acks.kt
+++ b/src/main/kotlin/io/github/nomisRev/kafka/publisher/Acks.kt
@@ -1,18 +1,19 @@
 package io.github.nomisRev.kafka.publisher
 
+import org.apache.kafka.clients.producer.ProducerConfig
+
 /**
  * The number of acknowledgments the producer requires the leader to have received before considering a request complete.
  * This controls the durability of records that are sent
  *
- * **Note:** that enabling idempotence requires this config value to be 'all'.
- * If conflicting configurations are set and idempotence is not explicitly enabled, idempotence is disabled.
+ * **NOTE:** Enabling idempotence requires this config value to be [All], otherwise [ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG] is ignored.
  */
 public enum class Acks(public val value: String) {
   /**
-   * If set to zero then the producer will not wait for any acknowledgment from the server at all.
+   * Using [Zero] the producer will not wait for any acknowledgment from the server at all.
    * The record will be immediately added to the socket buffer and considered sent.
    * No guarantee can be made that the server has received the record in this case,
-   * and the <code>retries</code> configuration will not take effect (as the client won't generally know of any failures).
+   * and [ProducerConfig.RETRIES_CONFIG] will not take effect (as the client won't generally know of any failures).
    * The offset given back for each record will always be set to `-1`
    */
   Zero("0"),
@@ -26,7 +27,7 @@ public enum class Acks(public val value: String) {
   /**
    * This means the leader will wait for the full set of in-sync replicas to acknowledge the record.
    * This guarantees that the record will not be lost as long as at least one in-sync replica remains alive.
-   * This is the strongest available guarantee. This is equivalent to the `acks=-1` setting.
+   * This is the strongest available guarantee. This is equivalent to the [MinusOne] setting.
    */
   All("all"),
 
diff --git a/src/main/kotlin/io/github/nomisRev/kafka/publisher/KafkaPublisher.kt b/src/main/kotlin/io/github/nomisRev/kafka/publisher/KafkaPublisher.kt
new file mode 100644
index 00000000..12af640c
--- /dev/null
+++ b/src/main/kotlin/io/github/nomisRev/kafka/publisher/KafkaPublisher.kt
@@ -0,0 +1,319 @@
+package io.github.nomisRev.kafka.publisher
+
+import io.github.nomisRev.kafka.publisher.DefaultKafkaPublisher.Companion.log
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CopyableThrowable
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.DEBUG_PROPERTY_NAME
+import kotlinx.coroutines.DEBUG_PROPERTY_VALUE_AUTO
+import kotlinx.coroutines.DEBUG_PROPERTY_VALUE_OFF
+import kotlinx.coroutines.DEBUG_PROPERTY_VALUE_ON
+import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.DelicateCoroutinesApi
+import kotlinx.coroutines.ExecutorCoroutineDispatcher
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.asCoroutineDispatcher
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.runInterruptible
+import kotlinx.coroutines.withContext
+import org.apache.kafka.clients.producer.KafkaProducer
+import org.apache.kafka.clients.producer.Producer
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.clients.producer.RecordMetadata
+import org.apache.kafka.common.Metric
+import org.apache.kafka.common.MetricName
+import org.apache.kafka.common.PartitionInfo
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+import java.time.Duration
+import java.util.concurrent.Executors
+import kotlin.coroutines.cancellation.CancellationException
+import kotlin.time.toJavaDuration
+
+/**
+ * Constructing a [KafkaPublisher] requires [PublisherSettings].
+ * Optionally you can provide a different way to how the [Producer] is created.
+ */
+fun <Key, Value> KafkaPublisher(
+  settings: PublisherSettings<Key, Value>,
+  configureProducer: suspend () -> Producer<Key, Value> =
+    { KafkaProducer(settings.properties(), settings.keySerializer, settings.valueSerializer) }
+): KafkaPublisher<Key, Value> =
+  DefaultKafkaPublisher(settings, configureProducer)
+
+/**
+ * A [KafkaPublisher] wraps an [Producer], so needs to be closed by [AutoCloseable].
+ * It has 1 main method, [publishScope] which creates a [PublishScope],
+ * and two suspending methods from the [Producer] [partitionsFor], and [metrics].
+ */
+interface KafkaPublisher<Key, Value> : AutoCloseable {
+
+  /**
+   * Create and run a [publishScope], which can [PublishScope.offer] and [PublishScope.publish] records to Kafka.
+   * It awaits all inflight offers to finish, and reports any errors.
+   *
+   * If the [block] fails, or one of the children of the created [CoroutineScope],
+   * than the exception is rethrown and the [PublishScope] gets cancelled.
+   *
+   * Just like [coroutineScope] it awaits all its children to finish.
+   *
+   * ```kotlin
+   * publisher.publishScope {
+   *   offer((1..10).map {
+   *     ProducerRecord(topic.name(), "$it", "msg-$it")
+   *   })
+   *   publish((11..20).map {
+   *     ProducerRecord(topic.name(), "$it", "msg-$it")
+   *   })
+   *   transaction {
+   *     // transaction { } compiler error: illegal to be called here
+   *     offer((21..30).map {
+   *       ProducerRecord(topic.name(), "$it", "msg-$it")
+   *     })
+   *     publish((31..40).map {
+   *       ProducerRecord(topic.name(), "$it", "msg-$it")
+   *     })
+   *   }// Waits until all offer finished in transaction, fails if any failed
+   *
+   *   // streaming
+   *   flow(1..100)
+   *     .onEach { delay(100.milliseconds) }
+   *     .map { ProducerRecord(topic.name(), "$it", "msg-$it") }
+   *     .collect { offer(it) }
+   * }
+   * ```
+   */
+  suspend fun <A> publishScope(block: suspend TransactionalScope<Key, Value>.() -> A): A
+
+  /** @see KafkaProducer.partitionsFor */
+  suspend fun partitionsFor(topic: String): List<PartitionInfo>
+
+  /** @see KafkaProducer.metrics */
+  suspend fun metrics(): Map<MetricName, Metric>
+
+}
+
+private class DefaultKafkaPublisher<Key, Value>(
+  val settings: PublisherSettings<Key, Value>,
+  createProducer: suspend () -> Producer<Key, Value>
+) : KafkaPublisher<Key, Value> {
+
+  val producerId = "reactor-kafka-sender-${System.identityHashCode(this)}"
+  val producerContext: ExecutorCoroutineDispatcher =
+    Executors.newScheduledThreadPool(1) { runnable ->
+      Thread(runnable, producerId).apply {
+        isDaemon = true
+      }
+    }.asCoroutineDispatcher()
+
+  @OptIn(DelicateCoroutinesApi::class)
+  val producer = GlobalScope.async(producerContext) {
+    createProducer().apply {
+      settings.producerListener.producerAdded(producerId, this)
+      if (settings.isTransactional()) {
+        log.info("Initializing transactions for producer {}", settings.transactionalId())
+        initTransactions()
+      }
+    }
+  }
+
+  override suspend fun partitionsFor(topic: String): List<PartitionInfo> {
+    val p = producer.await()
+    return runInterruptible(producerContext) {
+      p.partitionsFor(topic)
+    }
+  }
+
+
+  override suspend fun metrics(): Map<MetricName, Metric> {
+    val p = producer.await()
+    return runInterruptible(producerContext) {
+      p.metrics()
+    }
+  }
+
+  override suspend fun <A> publishScope(block: suspend TransactionalScope<Key, Value>.() -> A): A {
+    val token = Token()
+    return try {
+      coroutineScope {
+        val scope = DefaultProduceScope(
+          settings,
+          producer,
+          producerContext,
+          token,
+          this
+        )
+        block(scope)
+      }
+    } catch (e: ChildCancelScope) {
+      e.checkMyScope(token)
+    }
+  }
+
+  override fun close() = runBlocking {
+    listOf(
+      runCatching {
+        producer.await().close(
+          if (settings.closeTimeout.isInfinite()) Duration.ofMillis(Long.MAX_VALUE)
+          else settings.closeTimeout.toJavaDuration()
+        )
+      },
+      runCatching { settings.producerListener.producerRemoved(producerId, producer.await()) },
+      runCatching { producerContext.close() }
+    ).throwIfErrors()
+  }
+
+  companion object {
+    val log: Logger = LoggerFactory.getLogger(KafkaPublisher::class.java.name)
+  }
+}
+
+private fun Iterable<Result<Unit>>.throwIfErrors() {
+  fold<Result<Unit>, Throwable?>(null) { acc, result ->
+    acc?.apply {
+      result.exceptionOrNull()?.let { addSuppressed(it) }
+    } ?: result.exceptionOrNull()
+  }?.let { throw it }
+}
+
+private class Token
+
+private class DefaultProduceScope<Key, Value>(
+  val settings: PublisherSettings<Key, Value>,
+  val producer: Deferred<Producer<Key, Value>>,
+  val producerContext: ExecutorCoroutineDispatcher,
+  val token: Token,
+  scope: CoroutineScope
+) : TransactionalScope<Key, Value>, CoroutineScope by scope {
+  val parent: Job = requireNotNull(coroutineContext[Job]) { "Impossible, can only be called within coroutineScope" }
+
+  override suspend fun offer(record: ProducerRecord<Key, Value>) {
+    val p: Producer<Key, Value> = producer.await()
+    val child = Job(parent)
+    runInterruptible(producerContext) {
+      p.send(record) { _, exception ->
+        if (exception != null) parent.cancel(ChildCancelScope("Child failed", exception, token))
+        else child.complete()
+      }
+    }
+  }
+
+  override suspend fun publish(record: ProducerRecord<Key, Value>): RecordMetadata {
+    val p: Producer<Key, Value> = producer.await()
+    val promise = CompletableDeferred<RecordMetadata>()
+    runInterruptible(producerContext) {
+      p.send(record) { metadata, exception ->
+        if (exception != null) promise.completeExceptionally(exception)
+        else promise.complete(metadata)
+      }
+    }
+    return promise.await()
+  }
+
+  override suspend fun publishCatching(record: ProducerRecord<Key, Value>): Result<RecordMetadata> =
+    runCatching { publish(record) }.onFailure { settings.throwFatal(it) }
+
+  override suspend fun publishCatching(record: Iterable<ProducerRecord<Key, Value>>): Result<List<RecordMetadata>> =
+    runCatching {
+      coroutineScope {
+        record.map { async { publish(it) } }.awaitAll()
+      }
+    }.onFailure { settings.throwFatal(it) }
+
+  override suspend fun <A> transaction(block: suspend PublishScope<Key, Value>.() -> A): A {
+    val token = Token()
+    val p = producer.await()
+    withContext(producerContext) { p.beginTransaction() }
+    log.debug("Begin a new transaction for producer {}", settings.transactionalId())
+    return try {
+      coroutineScope {
+        val scope = DefaultProduceScope(
+          settings,
+          producer,
+          producerContext,
+          token,
+          this
+        )
+        block(scope)
+      }.also {
+        withContext(producerContext) { p.commitTransaction() }
+        log.debug("Commit current transaction for producer {}", settings.transactionalId())
+      }
+    } catch (e: Throwable) {
+      withContext(producerContext) { p.abortTransaction() }
+      log.debug("Abort current transaction for producer {}", settings.transactionalId())
+      if (e is ChildCancelScope) e.checkMyScope(token) else throw e
+    }
+  }
+
+  private fun <Key, Value> PublisherSettings<Key, Value>.throwFatal(t: Throwable): Unit =
+    if (
+    // Fatal JVM errors
+      t is VirtualMachineError ||
+      t is ThreadDeath ||
+      t is InterruptedException ||
+      t is LinkageError ||
+      // Fatal KotlinX error
+      t is CancellationException ||
+      // Fatal kafka errors
+      isFatal(t)
+    ) throw t else Unit
+}
+
+/*
+ * Marker type, that this is our cancellation signal.
+ * This allows us to check if it's our scope that is failing using [token].
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+private class ChildCancelScope(
+  message: String,
+  override val cause: Throwable,
+  @Transient @JvmField val token: Token,
+) : CancellationException(message), CopyableThrowable<ChildCancelScope> {
+  init {
+    initCause(cause)
+  }
+
+  /*
+   * In non-debug mode we don't want to have a stacktrace on every cancellation/close, parent job reference is enough.
+   * Stacktrace of JCE is not needed most of the time (e.g., it is not logged) and hurts performance.
+   */
+  override fun fillInStackTrace(): Throwable =
+    if (DEBUG) super.fillInStackTrace()
+    else apply {
+      stackTrace = emptyArray() // Prevent Android <= 6.0 bug, #1866
+    }
+
+  /* In non-debug mode we don't copy JCE for speed as it does not have the stack trace anyway. */
+  override fun createCopy(): ChildCancelScope? =
+    if (DEBUG) ChildCancelScope(message!!, cause, token)
+    else null
+
+  fun checkMyScope(scope: Token): Nothing =
+    when {
+      this.token === scope -> throw cause
+      else -> throw this
+    }
+}
+
+// Piggyback (copied) on KotlinX Coroutines DEBUG mechanism.
+// https://github.com/Kotlin/kotlinx.coroutines/blob/ed0cf7aa02b1266cb81e65e61b3a97b0e041a817/kotlinx-coroutines-core/jvm/src/Debug.kt#L70
+private val ASSERTIONS_ENABLED = ChildCancelScope::class.java.desiredAssertionStatus()
+
+private val DEBUG = try {
+  System.getProperty(DEBUG_PROPERTY_NAME)
+} catch (e: SecurityException) {
+  null
+}.let { value ->
+  when (value) {
+    DEBUG_PROPERTY_VALUE_AUTO, null -> ASSERTIONS_ENABLED
+    DEBUG_PROPERTY_VALUE_ON, "" -> true
+    DEBUG_PROPERTY_VALUE_OFF -> false
+    else -> error("System property '$DEBUG_PROPERTY_NAME' has unrecognized value '$value'")
+  }
+}
diff --git a/src/main/kotlin/io/github/nomisRev/kafka/publisher/PublisherScope.kt b/src/main/kotlin/io/github/nomisRev/kafka/publisher/PublisherScope.kt
new file mode 100644
index 00000000..a14fa040
--- /dev/null
+++ b/src/main/kotlin/io/github/nomisRev/kafka/publisher/PublisherScope.kt
@@ -0,0 +1,120 @@
+@file:JvmMultifileClass @file:JvmName("PublisherScope.kt")
+
+package io.github.nomisRev.kafka.publisher
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.coroutineScope
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.clients.producer.Producer
+import org.apache.kafka.clients.producer.RecordMetadata
+import org.apache.kafka.clients.producer.ProducerConfig
+import org.apache.kafka.common.errors.ProducerFencedException
+
+@DslMarker
+annotation class PublisherDSL
+
+/**
+ * The DSL, or receiver type, of [KafkaPublisher.publishScope] and [TransactionalScope.transaction].
+ *
+ * There are 2 main methods for sending recordings to kafka.
+ *  - [offer], this gives you highest throughput and should generally be preferred.
+ *  - [publish], this gives you less throughput, but waits on the delivery of the messages.
+ */
+@PublisherDSL
+interface PublishScope<Key, Value> : CoroutineScope {
+
+  /**
+   * Offer the [record] to Kafka, and immediately return.
+   * This methods should be prepared for highest throughput,
+   * if the [offer] fails it will cancel the [CoroutineScope] & [PublishScope].
+   *
+   * @param record to be offered to kafka
+   */
+  suspend fun offer(record: ProducerRecord<Key, Value>)
+
+  /**
+   * Publisher a [record] to Kafka, and suspends until acknowledged by kafka.
+   * This way of sending records to kafka results in a lower throughput.
+   *
+   * **IMPORTANT:** publish is slower than offer, if you need high throughput simply use [offer].
+   *   Cancelling doesn't cancel the [publish]/[Producer.send].
+   *
+   * @param record to be delivered to kafka
+   */
+  suspend fun publish(record: ProducerRecord<Key, Value>): RecordMetadata
+
+  /**
+   * Same as [offer], but for an [Iterable]] of [ProducerRecord].
+   * @see offer
+   */
+  suspend fun offer(records: Iterable<ProducerRecord<Key, Value>>) =
+    records.map { offer(it) }
+
+  /**
+   * Same as [publish], but for an [Iterable]] of [ProducerRecord].
+   * @see publish
+   */
+  suspend fun publish(record: Iterable<ProducerRecord<Key, Value>>): List<RecordMetadata> = coroutineScope {
+    record.map { async { publish(it) } }.awaitAll()
+  }
+
+  /** Alias for `runCatching`, and `publish` except rethrows fatal exceptions */
+  suspend fun publishCatching(record: ProducerRecord<Key, Value>): Result<RecordMetadata>
+
+  /**
+   * Catch first failure of [publish], except fatal exceptions.
+   * Alias for `runCatching`, `publish`, and `awaitAll`, except rethrows fatal exceptions
+   */
+  suspend fun publishCatching(record: Iterable<ProducerRecord<Key, Value>>): Result<List<RecordMetadata>>
+}
+
+/**
+ * The DSL, or receiver type, of [KafkaPublisher.publishScope],
+ * it allows for publisher configured with [ProducerConfig.TRANSACTIONAL_ID_CONFIG],
+ * and [ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG] set to `true` to work transitionally.
+ * Only one publisher with the same [ProducerConfig.TRANSACTIONAL_ID_CONFIG] can run at a given time.
+ * An expired producer, still works but will throw [ProducerFencedException] if attempt to make a transaction again.
+ *
+ * In addition to [PublishScope] 2 main methods, a [TransactionalScope] also has the [transaction] method.
+ *
+ * It creates a **new** nested [PublishScope], and runs the lambda with the new scope as receiver.
+ *
+ * If the lambda succeeds, and the children of [Job] have completed,
+ * it automatically commits the [transaction].
+ *
+ * In case the lambda fails, or any children of this [Job] fail,
+ * then it'll be rethrown from this block and the transaction will be aborted.
+ */
+@PublisherDSL
+interface TransactionalScope<Key, Value> : PublishScope<Key, Value> {
+
+  /**
+   * Create and run a [transaction], which can [PublishScope.offer] and [PublishScope.publish] records to Kafka.
+   * It awaits all inflight offers to finish, when successfully it commits the transaction and returns.
+   * In case of failure
+   *
+   * If the [block] fails, or one of the children of the created [CoroutineScope],
+   * then it aborts the transaction and the exception is rethrown and the [PublishScope] gets cancelled.
+   *
+   * Just like [coroutineScope] it awaits all its children to finish.
+   *
+   * ```kotlin
+   * publisher.publishScope {
+   *   transaction {
+   *     // transaction { } compiler error: illegal to be called here
+   *     offer((21..30).map {
+   *       ProducerRecord(topic.name(), "$it", "msg-$it")
+   *     })
+   *     publish((31..40).map {
+   *       ProducerRecord(topic.name(), "$it", "msg-$it")
+   *     })
+   *   }// Waits until all offer finished in transaction, fails if any failed
+   * }
+   * ```
+   */
+  suspend fun <A> transaction(block: suspend PublishScope<Key, Value>.() -> A): A
+}
diff --git a/src/main/kotlin/io/github/nomisRev/kafka/publisher/PublisherSettings.kt b/src/main/kotlin/io/github/nomisRev/kafka/publisher/PublisherSettings.kt
index ee962335..56f045fc 100644
--- a/src/main/kotlin/io/github/nomisRev/kafka/publisher/PublisherSettings.kt
+++ b/src/main/kotlin/io/github/nomisRev/kafka/publisher/PublisherSettings.kt
@@ -1,33 +1,34 @@
 package io.github.nomisRev.kafka.publisher
 
 import io.github.nomisRev.kafka.NothingSerializer
+import io.github.nomisRev.kafka.publisher.PublisherSettings.ProducerListener
 import io.github.nomisRev.kafka.receiver.isPosNonZero
-import io.github.nomisRev.kafka.publisher.PublisherOptions.ProducerListener
-import kotlinx.coroutines.channels.Channel
 import org.apache.kafka.clients.producer.Producer
 import org.apache.kafka.clients.producer.ProducerConfig
+import org.apache.kafka.common.errors.AuthenticationException
+import org.apache.kafka.common.errors.ProducerFencedException
 import org.apache.kafka.common.serialization.Serializer
 import java.util.Properties
 import kotlin.time.Duration
 
 /**
+ * Typed setting to create a [KafkaPublisher], enforces the required parameters and leaves the rest as [Properties].
+ *
  * @param bootstrapServers list of host/port pairs to use for establishing the initial connection to the Kafka cluster. Should be comma separated.
  * @param keySerializer the [Serializer] to use to serialize the [Key] when sending messages to kafka.
  * @param valueSerializer the [Serializer] to use to serialize the [Value] when sending messages to kafka.
  * @param acknowledgments configuration to use.
  * @param closeTimeout the timeout when closing the created underlying [Producer], default [Duration.INFINITE].
- * @param maxInFlight the maximum number of in-flight records that are fetched from the outbound record publisher while acknowledgements are pending. Default [Channel.BUFFERED].
- * @param stopOnError if a send operation should be terminated when an error is encountered. If set to false, send is attempted for all records in a sequence even if send of one of the records fails with a non-fatal exception.
  * @param producerListener listener that is called whenever a [Producer] is added, and removed.
  */
-data class PublisherOptions<Key, Value>(
+data class PublisherSettings<Key, Value>(
   val bootstrapServers: String,
   val keySerializer: Serializer<Key>,
   val valueSerializer: Serializer<Value>,
   val acknowledgments: Acks = Acks.One,
   val closeTimeout: Duration = Duration.INFINITE,
-  val maxInFlight: Int = Channel.BUFFERED,
-  val stopOnError: Boolean = true,
+  val isFatal: (t: Throwable) -> Boolean =
+    { it is AuthenticationException || it is ProducerFencedException },
   val producerListener: ProducerListener = NoOpProducerListener,
   val properties: Properties = Properties(),
 ) {
@@ -44,6 +45,18 @@ data class PublisherOptions<Key, Value>(
     putAll(properties)
   }
 
+  fun transactionalId(): String? =
+    properties[ProducerConfig.TRANSACTIONAL_ID_CONFIG] as? String
+
+  /**
+   * [KafkaPublisher] created from this options will be transactional if a transactional id is
+   * configured using [ProducerConfig.TRANSACTIONAL_ID_CONFIG]. If transactional,
+   * [Producer.initTransactions] is invoked on the producer to initialize
+   * transactions before any operations are performed on the publisher.
+   */
+  fun isTransactional(): Boolean =
+    !transactionalId().isNullOrBlank()
+
   /** Called whenever a [Producer] is added or removed. */
   interface ProducerListener {
     /**
@@ -62,25 +75,24 @@ data class PublisherOptions<Key, Value>(
   }
 }
 
-/** Alternative constructor for [PublisherOptions] without a key */
+/** Alternative constructor for [PublisherSettings] without a key */
 public fun <Value> PublisherOptions(
   bootstrapServers: String,
   valueSerializer: Serializer<Value>,
   acknowledgments: Acks = Acks.One,
   closeTimeout: Duration = Duration.INFINITE,
-  maxInFlight: Int = Channel.BUFFERED,
-  stopOnError: Boolean = true,
+  isFatal: (t: Throwable) -> Boolean =
+    { it is AuthenticationException || it is ProducerFencedException },
   producerListener: ProducerListener = NoOpProducerListener,
   properties: Properties = Properties(),
-): PublisherOptions<Nothing, Value> =
-  PublisherOptions(
+): PublisherSettings<Nothing, Value> =
+  PublisherSettings(
     bootstrapServers,
     NothingSerializer,
     valueSerializer,
     acknowledgments,
     closeTimeout,
-    maxInFlight,
-    stopOnError,
+    isFatal,
     producerListener,
     properties
   )
diff --git a/src/test/kotlin/io/github/nomisrev/kafka/KafkaSpec.kt b/src/test/kotlin/io/github/nomisrev/kafka/KafkaSpec.kt
index c1649afb..011f98b7 100644
--- a/src/test/kotlin/io/github/nomisrev/kafka/KafkaSpec.kt
+++ b/src/test/kotlin/io/github/nomisrev/kafka/KafkaSpec.kt
@@ -3,42 +3,63 @@ package io.github.nomisrev.kafka
 import io.github.nomisRev.kafka.Admin
 import io.github.nomisRev.kafka.AdminSettings
 import io.github.nomisRev.kafka.receiver.AutoOffsetReset
-import io.github.nomisRev.kafka.ProducerSettings
 import io.github.nomisRev.kafka.createTopic
 import io.github.nomisRev.kafka.deleteTopic
 import io.github.nomisRev.kafka.describeTopic
-import io.github.nomisRev.kafka.produce
+import io.github.nomisRev.kafka.publisher.Acks
+import io.github.nomisRev.kafka.publisher.KafkaPublisher
+import io.github.nomisRev.kafka.publisher.PublisherSettings
 import io.github.nomisRev.kafka.receiver.KafkaReceiver
 import io.github.nomisRev.kafka.receiver.ReceiverSettings
+import io.kotest.assertions.assertSoftly
+import io.kotest.assertions.async.shouldTimeout
 import io.kotest.core.spec.style.StringSpec
-import kotlinx.coroutines.flow.asFlow
-import kotlinx.coroutines.flow.collect
+import io.kotest.matchers.shouldBe
+import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.take
+import kotlinx.coroutines.flow.toList
 import org.apache.kafka.clients.admin.Admin
 import org.apache.kafka.clients.admin.AdminClientConfig
 import org.apache.kafka.clients.admin.NewTopic
+import org.apache.kafka.clients.consumer.ConsumerGroupMetadata
+import org.apache.kafka.clients.consumer.OffsetAndMetadata
+import org.apache.kafka.clients.producer.Callback
+import org.apache.kafka.clients.producer.KafkaProducer
+import org.apache.kafka.clients.producer.Producer
 import org.apache.kafka.clients.producer.ProducerConfig
 import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.clients.producer.RecordMetadata
+import org.apache.kafka.common.Metric
+import org.apache.kafka.common.MetricName
+import org.apache.kafka.common.PartitionInfo
 import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.serialization.StringDeserializer
 import org.apache.kafka.common.serialization.StringSerializer
 import org.testcontainers.containers.KafkaContainer
 import org.testcontainers.utility.DockerImageName
+import java.time.Duration
 import java.util.Properties
 import java.util.UUID
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.Executors
+import java.util.concurrent.Future
+import java.util.concurrent.TimeUnit
+import kotlin.time.Duration.Companion.milliseconds
 import kotlin.time.Duration.Companion.seconds
 
 abstract class KafkaSpec(body: KafkaSpec.() -> Unit = {}) : StringSpec() {
   init {
     body()
   }
-  
+
   private val transactionTimeoutInterval = 1.seconds
   private val consumerPollingTimeout = 1.seconds
-  
+  val boom = RuntimeException("Boom!")
+
   private val kafkaImage: DockerImageName =
     DockerImageName.parse("confluentinc/cp-kafka:latest")
-  
+
   private val container: KafkaContainer = autoClose(
     KafkaContainer(kafkaImage)
       .withExposedPorts(9092, 9093)
@@ -55,20 +76,20 @@ abstract class KafkaSpec(body: KafkaSpec.() -> Unit = {}) : StringSpec() {
       .withReuse(true)
       .also { container -> container.start() }
   )
-  
+
   private fun adminProperties(): Properties = Properties().apply {
     put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, container.bootstrapServers)
     put(AdminClientConfig.CLIENT_ID_CONFIG, "test-kafka-admin-client")
     put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "10000")
     put(AdminClientConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG, "10000")
   }
-  
+
   fun adminSettings(): AdminSettings =
     AdminSettings(container.bootstrapServers, adminProperties())
-  
+
   inline fun <A> admin(body: Admin.() -> A): A =
     Admin(adminSettings()).use(body)
-  
+
   fun receiverSetting(): ReceiverSettings<String, String> =
     ReceiverSettings(
       bootstrapServers = container.bootstrapServers,
@@ -78,25 +99,31 @@ abstract class KafkaSpec(body: KafkaSpec.() -> Unit = {}) : StringSpec() {
       autoOffsetReset = AutoOffsetReset.Earliest,
       pollTimeout = consumerPollingTimeout
     )
-  
-  fun producerSettings(): ProducerSettings<String, String> =
-    ProducerSettings(
+
+  fun publisherSettings(
+    acknowledgments: Acks = Acks.One,
+    properties: Properties = Properties()
+  ): PublisherSettings<String, String> =
+    PublisherSettings(
       bootstrapServers = container.bootstrapServers,
-      keyDeserializer = StringSerializer(),
-      valueDeserializer = StringSerializer(),
-      other = Properties().apply {
-        put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, container.bootstrapServers)
+      keySerializer = StringSerializer(),
+      valueSerializer = StringSerializer(),
+      acknowledgments = acknowledgments,
+      properties = properties.apply {
         put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10000.toString())
         put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 1000.toString())
       }
     )
-  
+
+  val producer = KafkaProducer<String, String>(publisherSettings().properties())
+  val publisher = autoClose(KafkaPublisher(publisherSettings()) { producer })
+
   private fun nextTopicName(): String =
     "topic-${UUID.randomUUID()}"
-  
+
   suspend fun <A> withTopic(
     topicConfig: Map<String, String> = emptyMap(),
-    partitions: Int = 1,
+    partitions: Int = 4,
     replicationFactor: Short = 1,
     action: suspend Admin.(NewTopic) -> A,
   ): A {
@@ -106,34 +133,33 @@ abstract class KafkaSpec(body: KafkaSpec.() -> Unit = {}) : StringSpec() {
       try {
         action(topic)
       } finally {
+        topic.shouldBeEmpty()
         deleteTopic(topic.name())
       }
     }
   }
-  
+
   @JvmName("publishPairsToKafka")
   suspend fun publishToKafka(
     topic: NewTopic,
     messages: Iterable<Pair<String, String>>,
   ): Unit =
-    messages.asFlow()
-      .map { (key, value) ->
-        ProducerRecord(topic.name(), key, value)
-      }.produce(producerSettings())
-      .collect()
-  
+    publishToKafka(messages.map { (key, value) ->
+      ProducerRecord(topic.name(), key, value)
+    })
+
   suspend fun publishToKafka(messages: Iterable<ProducerRecord<String, String>>): Unit =
-    messages.asFlow()
-      .produce(producerSettings())
-      .collect()
-  
+    publisher.publishScope {
+      offer(messages)
+    }
+
   suspend fun <K, V> KafkaReceiver<K, V>.committedCount(topic: String): Long =
     admin {
       val description = requireNotNull(describeTopic(topic)) { "Topic $topic not found" }
       val topicPartitions = description.partitions().map {
         TopicPartition(topic, it.partition())
       }.toSet()
-      
+
       withConsumer {
         committed(topicPartitions)
           .mapNotNull { (_, offset) ->
@@ -141,4 +167,127 @@ abstract class KafkaSpec(body: KafkaSpec.() -> Unit = {}) : StringSpec() {
           }.sum()
       }
     }
+
+  @JvmName("shouldHaveAllRecords")
+  suspend fun NewTopic.shouldHaveRecords(
+    records: Iterable<Iterable<ProducerRecord<String, String>>>
+  ) {
+    val expected =
+      records.flatten().groupBy({ it.partition() }) { it.value() }.mapValues { it.value.toSet() }
+    KafkaReceiver(receiverSetting())
+      .receive(name())
+      .map { record ->
+        record.also { record.offset.acknowledge() }
+      }
+      .take(records.flatten().size)
+      .toList()
+      .groupBy({ it.partition() }) { it.value() }
+      .mapValues { it.value.toSet() } shouldBe expected
+  }
+
+  suspend fun NewTopic.shouldHaveRecords(records: Iterable<ProducerRecord<String, String>>) {
+    KafkaReceiver(receiverSetting())
+      .receive(name())
+      .map { record ->
+        record
+          .also { record.offset.acknowledge() }
+      }
+      .take(records.toList().size)
+      .toList()
+      .groupBy({ it.partition() }) { it.value() } shouldBe records.groupBy({ it.partition() }) { it.value() }
+  }
+
+  suspend fun NewTopic.shouldBeEmpty() {
+    shouldTimeout(100.milliseconds) {
+      KafkaReceiver(receiverSetting())
+        .receive(name())
+        .take(1)
+        .toList()
+    }
+  }
+
+  suspend fun NewTopic.shouldHaveRecord(records: ProducerRecord<String, String>) {
+    assertSoftly {
+      KafkaReceiver(receiverSetting())
+        .receive(name())
+        .map {
+          it.apply { offset.acknowledge() }
+        }.take(1)
+        .map { it.value() }
+        .toList() shouldBe listOf(records.value())
+      shouldBeEmpty()
+    }
+  }
+
+  suspend fun topicWithSingleMessage(topic: NewTopic, record: ProducerRecord<String, String>) =
+    KafkaReceiver(receiverSetting())
+      .receive(topic.name())
+      .map {
+        it.apply { offset.acknowledge() }
+      }.first().value() shouldBe record.value()
+
+  suspend fun topicShouldBeEmpty(topic: NewTopic) =
+    shouldTimeout(1.seconds) {
+      KafkaReceiver(receiverSetting())
+        .receive(topic.name())
+        .take(1)
+        .toList()
+    }
+
+  fun stubProducer(failOnNumber: Int? = null): suspend () -> Producer<String, String> = suspend {
+    object : Producer<String, String> {
+      override fun close() {}
+
+      override fun close(timeout: Duration?) {}
+
+      override fun initTransactions() =
+        producer.initTransactions()
+
+      override fun beginTransaction() =
+        producer.beginTransaction()
+
+      @Suppress("DEPRECATION")
+      @Deprecated("Deprecated in Java")
+      override fun sendOffsetsToTransaction(
+        offsets: MutableMap<TopicPartition, OffsetAndMetadata>?,
+        consumerGroupId: String?
+      ) = producer.sendOffsetsToTransaction(offsets, consumerGroupId)
+
+      override fun sendOffsetsToTransaction(
+        offsets: MutableMap<TopicPartition, OffsetAndMetadata>?,
+        groupMetadata: ConsumerGroupMetadata?
+      ) = producer.sendOffsetsToTransaction(offsets, groupMetadata)
+
+      override fun commitTransaction() =
+        producer.commitTransaction()
+
+      override fun abortTransaction() =
+        producer.abortTransaction()
+
+      override fun flush() =
+        producer.flush()
+
+      override fun partitionsFor(topic: String?): MutableList<PartitionInfo> =
+        producer.partitionsFor(topic)
+
+      override fun metrics(): MutableMap<MetricName, out Metric> =
+        producer.metrics()
+
+      override fun send(record: ProducerRecord<String, String>, callback: Callback): Future<RecordMetadata> =
+        if (failOnNumber != null && record.key() == failOnNumber.toString()) {
+          Executors.newScheduledThreadPool(1).schedule(
+            {
+              callback.onCompletion(null, boom)
+            },
+            50,
+            TimeUnit.MILLISECONDS
+          )
+
+          CompletableFuture.supplyAsync { throw AssertionError("Should never be called") }
+        } else producer.send(record, callback)
+
+      override fun send(record: ProducerRecord<String, String>): Future<RecordMetadata> =
+        producer.send(record)
+    }
+  }
 }
diff --git a/src/test/kotlin/io/github/nomisrev/kafka/Predef.kt b/src/test/kotlin/io/github/nomisrev/kafka/Predef.kt
index 6b5cdc2d..3c01de3d 100644
--- a/src/test/kotlin/io/github/nomisrev/kafka/Predef.kt
+++ b/src/test/kotlin/io/github/nomisrev/kafka/Predef.kt
@@ -1,5 +1,9 @@
 package io.github.nomisrev.kafka
 
+import io.kotest.assertions.fail
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.map
 
@@ -10,4 +14,15 @@ inline fun <A, B> Flow<A>.mapIndexed(
   return map { value ->
     transform(index++, value)
   }
+}
+
+suspend fun CoroutineScope.shouldCancel(block: suspend () -> Unit): CancellationException {
+  val cancel = CompletableDeferred<CancellationException>()
+  try {
+    block()
+    fail("Expected to be cancellable, but wasn't")
+  } catch (e: CancellationException) {
+    cancel.complete(e)
+  }
+  return cancel.await()
 }
\ No newline at end of file
diff --git a/src/test/kotlin/io/github/nomisrev/kafka/publisher/KafkaPublisherSpec.kt b/src/test/kotlin/io/github/nomisrev/kafka/publisher/KafkaPublisherSpec.kt
new file mode 100644
index 00000000..d127bb6f
--- /dev/null
+++ b/src/test/kotlin/io/github/nomisrev/kafka/publisher/KafkaPublisherSpec.kt
@@ -0,0 +1,310 @@
+package io.github.nomisrev.kafka.publisher
+
+import io.github.nomisRev.kafka.publisher.Acks
+import io.github.nomisRev.kafka.publisher.KafkaPublisher
+import io.github.nomisrev.kafka.KafkaSpec
+import io.kotest.assertions.throwables.shouldThrow
+import io.kotest.matchers.shouldBe
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineStart.UNDISPATCHED
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.awaitCancellation
+import kotlinx.coroutines.launch
+import org.apache.kafka.clients.admin.NewTopic
+import org.apache.kafka.clients.producer.ProducerConfig
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.common.errors.ProducerFencedException
+import java.util.Properties
+
+class KafkaPublisherSpec : KafkaSpec({
+
+  "All offered messages are received" {
+    withTopic { topic ->
+      val count = 3
+      val records = (0..count).map {
+        topic.createProducerRecord(it)
+      }
+      publisher.publishScope {
+        offer(records)
+      }
+
+      topic.shouldHaveRecords(records)
+    }
+  }
+
+  "Can receive all messages that were published on the right partitions" {
+    withTopic { topic ->
+      val count = 3
+      val records = (0..count).map {
+        topic.createProducerRecord(it)
+      }
+      publisher.publishScope {
+        publish(records)
+      }
+
+      topic.shouldHaveRecords(records)
+    }
+  }
+
+  "A failure in a produce block, rethrows the error" {
+    withTopic { topic ->
+      val record = topic.createProducerRecord(0)
+
+      shouldThrow<RuntimeException> {
+        publisher.publishScope<Unit> {
+          offer(record)
+          throw boom
+        }
+      } shouldBe boom
+
+      topic.shouldHaveRecord(record)
+    }
+  }
+
+  "A failure in a produce block with a concurrent launch cancels the launch, rethrows the error" {
+    withTopic { _ ->
+      val cancelSignal = CompletableDeferred<CancellationException>()
+      shouldThrow<RuntimeException> {
+        publisher.publishScope<Unit> {
+          launch(start = UNDISPATCHED) {
+            try {
+              awaitCancellation()
+            } catch (e: CancellationException) {
+              cancelSignal.complete(e)
+              throw e
+            }
+          }
+          throw boom
+        }
+      } shouldBe boom
+      cancelSignal.await()
+    }
+  }
+
+  "A failed offer is rethrown at the end" {
+    withTopic { topic ->
+      val record = topic.createProducerRecord(0)
+
+      shouldThrow<RuntimeException> {
+        KafkaPublisher(publisherSettings(), stubProducer(failOnNumber = 0)).publishScope {
+          offer(record)
+        }
+      } shouldBe boom
+    }
+  }
+
+  "An async failure is rethrow at the end" {
+    withTopic { topic ->
+      val count = 3
+      val records: List<ProducerRecord<String, String>> = (0..count).map {
+        topic.createProducerRecord(it)
+      }
+      shouldThrow<RuntimeException> {
+        publisher.publishScope {
+          publish(records)
+          launch { throw boom }
+        }
+      } shouldBe boom
+
+      topic.shouldHaveRecords(records)
+    }
+  }
+
+  "A failure of a sendAwait can be caught in the block" {
+    withTopic { topic ->
+      val record0 = topic.createProducerRecord(0)
+      val record1 = topic.createProducerRecord(1)
+
+      KafkaPublisher(publisherSettings(), stubProducer(failOnNumber = 0)).use {
+        it.publishScope {
+          publishCatching(record0)
+          offer(record1)
+        }
+      }
+
+      topic.shouldHaveRecord(record1)
+    }
+  }
+
+  "concurrent publishing" {
+    withTopic { topic ->
+      val count = 4
+      val records =
+        (0..<count).map { base ->
+          (base..base + count).map { topic.createProducerRecord(it) }
+        }
+
+      publisher.publishScope {
+        listOf(
+          async { offer(records[0]) },
+          async { offer(records[1]) },
+          async { publish(records[2]) },
+          async { publish(records[3]) }
+        ).awaitAll()
+      }
+
+      topic.shouldHaveRecords(records)
+    }
+  }
+
+  "transaction an receive all messages that were published on the right partitions" {
+    withTopic { topic ->
+      val count = 3
+      val records = (0..count).map {
+        topic.createProducerRecord(it)
+      }
+      val settings = publisherSettings(
+        acknowledgments = Acks.All,
+        properties = Properties().apply {
+          put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, testCase.name.testName)
+          put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true")
+        }
+      )
+      KafkaPublisher(settings).use {
+        it.publishScope {
+          transaction {
+            offer(records)
+          }
+        }
+      }
+
+      topic.shouldHaveRecords(records)
+    }
+  }
+
+  "A failure in a transaction aborts the transaction" {
+    withTopic { topic ->
+      val count = 3
+      val records = (0..count).map {
+        topic.createProducerRecord(it)
+      }
+      val settings = publisherSettings(
+        acknowledgments = Acks.All,
+        properties = Properties().apply {
+          put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, testCase.name.testName)
+          put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true")
+        }
+      )
+      shouldThrow<RuntimeException> {
+        KafkaPublisher(settings).use {
+          it.publishScope<Unit> {
+            transaction {
+              publish(records)
+              throw boom
+            }
+          }
+        }
+      } shouldBe boom
+
+      topic.shouldBeEmpty()
+    }
+  }
+
+  "An async failure in a transaction aborts the transaction" {
+    withTopic { topic ->
+      val count = 3
+      val records = (0..count).map {
+        topic.createProducerRecord(it)
+      }
+      val settings = publisherSettings(
+        acknowledgments = Acks.All,
+        properties = Properties().apply {
+          put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, testCase.name.testName)
+          put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true")
+        }
+      )
+      shouldThrow<RuntimeException> {
+        KafkaPublisher(settings).use {
+          it.publishScope {
+            transaction {
+              offer(records)
+              launch { throw boom }
+            }
+          }
+        }
+      } shouldBe boom
+
+      topic.shouldBeEmpty()
+    }
+  }
+
+  "transaction - concurrent publishing" {
+    withTopic { topic ->
+      val count = 4
+      val records =
+        (0..<count).map { base ->
+          (base..base + count).map { topic.createProducerRecord(it) }
+        }
+
+      val settings = publisherSettings(
+        acknowledgments = Acks.All,
+        properties = Properties().apply {
+          put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, testCase.name.testName)
+          put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true")
+        }
+      )
+
+      KafkaPublisher(settings).use {
+        it.publishScope {
+          transaction {
+            listOf(
+              async { offer(records[0]) },
+              async { offer(records[1]) },
+              async { publish(records[2]) },
+              async { publish(records[3]) }
+            ).awaitAll()
+          }
+        }
+      }
+
+      topic.shouldHaveRecords(records)
+    }
+  }
+
+  "Only one KafkaProducer can have transactional.id at the same time, resulting ProducerFencedException is fatal" {
+    withTopic { topic ->
+      val settings = publisherSettings(
+        acknowledgments = Acks.All,
+        properties = Properties().apply {
+          put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, testCase.name.testName)
+          put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true")
+        }
+      )
+      val records1 = (0..4).map { topic.createProducerRecord(it) }
+      val publisher1 = KafkaPublisher(settings)
+      publisher1.publishScope {
+        transaction {
+          publish(records1)
+        }
+      }
+
+      val records2 = (5..9).map { topic.createProducerRecord(it) }
+      val publisher2 = KafkaPublisher(settings)
+      publisher2.publishScope {
+        transaction {
+          publish(records2)
+        }
+      }
+
+      // publisher1 was previous transactional.id, will result in fatal ProducerFencedException
+      val records3 = (10..14).map { topic.createProducerRecord(it) }
+      shouldThrow<ProducerFencedException> {
+        publisher1.publishScope {
+          transaction {
+            publishCatching(records3)
+          }
+        }
+      }
+
+      // Due to ProducerFencedException, only records1 and records2 are received
+      topic.shouldHaveRecords(records1 + records2)
+    }
+  }
+})
+
+fun NewTopic.createProducerRecord(index: Int, partitions: Int = 4): ProducerRecord<String, String> {
+  val partition: Int = index % partitions
+  return ProducerRecord<String, String>(name(), partition, "$index", "Message $index")
+}