diff --git a/docs/content/docs/deployment/module.md b/docs/content/docs/deployment/module.md
index d0854f7..895dd10 100644
--- a/docs/content/docs/deployment/module.md
+++ b/docs/content/docs/deployment/module.md
@@ -63,7 +63,7 @@
           address: kafka-broker:9092
           deliverySemantic:
             type: exactly-once
-            transactionTimeoutMillis: 100000
+            transactionTimeout: 15min
 ```
 
 ## Endpoint Definition
@@ -281,7 +281,7 @@
 An egress identifier, similar to a function type, uniquely identifies an egress.
 
 The spec defines the details of how to connect to the external system, which is specific to each individual I/O module.
-Each identifier-spec pair is bound to the system inside an stateful function module.
+Each identifier-spec pair is bound to the system inside a stateful function module.
 
 See [IO Modules]{{< ref "docs/io-module/overview" >}} for more information on configuring an egress. 
 
diff --git a/docs/content/docs/io-module/apache-kafka.md b/docs/content/docs/io-module/apache-kafka.md
index e0b4653..f331dc2 100644
--- a/docs/content/docs/io-module/apache-kafka.md
+++ b/docs/content/docs/io-module/apache-kafka.md
@@ -140,7 +140,7 @@
          address: kafka-broker:9092
          deliverySemantic:
            type: exactly-once
-           transactionTimeoutMillis: 100000
+           transactionTimeout: 15min
          properties:
            - foo.config: bar
 ```
@@ -177,7 +177,7 @@
 ```yaml
 deliverySemantic:
   type: exactly-once
-  transactionTimeoutMillis: 900000 # 15 min
+  transactionTimeoutMillis: 15min
 ```
 
 ### Writing To Kafka
diff --git a/statefun-e2e-tests/statefun-exactly-once-remote-e2e/src/test/resources/remote-module/module.yaml b/statefun-e2e-tests/statefun-exactly-once-remote-e2e/src/test/resources/remote-module/module.yaml
index c9704b1..de67146 100644
--- a/statefun-e2e-tests/statefun-exactly-once-remote-e2e/src/test/resources/remote-module/module.yaml
+++ b/statefun-e2e-tests/statefun-exactly-once-remote-e2e/src/test/resources/remote-module/module.yaml
@@ -51,4 +51,4 @@
             address: kafka-broker:9092
             deliverySemantic:
               type: exactly-once
-              transactionTimeoutMillis: 900000
+              transactionTimeout: 15min
diff --git a/statefun-flink/statefun-flink-common/src/main/java/org/apache/flink/statefun/flink/common/json/Selectors.java b/statefun-flink/statefun-flink-common/src/main/java/org/apache/flink/statefun/flink/common/json/Selectors.java
index 755c3a9..a5b9d2c 100644
--- a/statefun-flink/statefun-flink-common/src/main/java/org/apache/flink/statefun/flink/common/json/Selectors.java
+++ b/statefun-flink/statefun-flink-common/src/main/java/org/apache/flink/statefun/flink/common/json/Selectors.java
@@ -17,11 +17,13 @@
  */
 package org.apache.flink.statefun.flink.common.json;
 
+import java.time.Duration;
 import java.util.*;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonPointer;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.flink.util.TimeUtils;
 
 public final class Selectors {
 
@@ -44,6 +46,36 @@
     return Optional.of(node.asText());
   }
 
+  public static Duration durationAt(JsonNode node, JsonPointer pointer) {
+    node = dereference(node, pointer);
+    if (!node.isTextual()) {
+      throw new WrongTypeException(pointer, "not a duration");
+    }
+
+    try {
+      return TimeUtils.parseDuration(node.asText());
+    } catch (IllegalArgumentException ignore) {
+      throw new WrongTypeException(pointer, "not a duration");
+    }
+  }
+
+  public static Optional<Duration> optionalDurationAt(JsonNode node, JsonPointer pointer) {
+    node = node.at(pointer);
+    if (node.isMissingNode()) {
+      return Optional.empty();
+    }
+    if (!node.isTextual()) {
+      throw new WrongTypeException(pointer, "not a duration");
+    }
+
+    try {
+      Duration duration = TimeUtils.parseDuration(node.asText());
+      return Optional.of(duration);
+    } catch (IllegalArgumentException ignore) {
+      throw new WrongTypeException(pointer, "not a duration");
+    }
+  }
+
   public static int integerAt(JsonNode node, JsonPointer pointer) {
     node = dereference(node, pointer);
     if (!node.isInt()) {
@@ -60,6 +92,18 @@
     return node.asLong();
   }
 
+  public static OptionalLong optionalLongAt(JsonNode node, JsonPointer pointer) {
+    node = node.at(pointer);
+    if (node.isMissingNode()) {
+      return OptionalLong.empty();
+    }
+
+    if (!node.isLong() && !node.isInt()) {
+      throw new WrongTypeException(pointer, "not a long");
+    }
+    return OptionalLong.of(node.asLong());
+  }
+
   public static OptionalInt optionalIntegerAt(JsonNode node, JsonPointer pointer) {
     node = node.at(pointer);
     if (node.isMissingNode()) {
diff --git a/statefun-flink/statefun-flink-common/src/test/java/org/apache/flink/statefun/flink/common/json/SelectorsTest.java b/statefun-flink/statefun-flink-common/src/test/java/org/apache/flink/statefun/flink/common/json/SelectorsTest.java
index e1f7b74..ced6db4 100644
--- a/statefun-flink/statefun-flink-common/src/test/java/org/apache/flink/statefun/flink/common/json/SelectorsTest.java
+++ b/statefun-flink/statefun-flink-common/src/test/java/org/apache/flink/statefun/flink/common/json/SelectorsTest.java
@@ -23,9 +23,11 @@
 import static org.hamcrest.Matchers.hasEntry;
 import static org.junit.Assert.assertThat;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.OptionalLong;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonPointer;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
@@ -79,6 +81,26 @@
   }
 
   @Test
+  public void durationAt() {
+    ObjectNode node = new ObjectNode(mapper.getNodeFactory());
+    node.put("foo", "30s");
+
+    Duration value = Selectors.durationAt(node, FOO_FIELD);
+
+    assertThat(value, is(Duration.ofSeconds(30)));
+  }
+
+  @Test
+  public void optionalDurationAt() {
+    ObjectNode node = new ObjectNode(mapper.getNodeFactory());
+    node.put("foo", "30s");
+
+    Optional<Duration> value = Selectors.optionalDurationAt(node, FOO_FIELD);
+
+    assertThat(value, is(Optional.of(Duration.ofSeconds(30))));
+  }
+
+  @Test
   public void longAt() {
     ObjectNode node = newObject();
     node.put("foo", 100_000L);
@@ -89,6 +111,16 @@
   }
 
   @Test
+  public void optionalLongAt() {
+    ObjectNode node = newObject();
+    node.put("foo", 100_000L);
+
+    OptionalLong value = Selectors.optionalLongAt(node, FOO_FIELD);
+
+    assertThat(value, is(OptionalLong.of(100_000L)));
+  }
+
+  @Test
   public void listAt() {
     ObjectNode node = newObject();
     node.putArray("foo").add(1).add(2).add(3);
diff --git a/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kafka/KafkaEgressSpecJsonParser.java b/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kafka/KafkaEgressSpecJsonParser.java
index 81f2947..b357fe9 100644
--- a/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kafka/KafkaEgressSpecJsonParser.java
+++ b/statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kafka/KafkaEgressSpecJsonParser.java
@@ -22,6 +22,7 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
+import java.util.OptionalLong;
 import java.util.Properties;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonPointer;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
@@ -40,9 +41,14 @@
       JsonPointer.compile("/egress/spec/deliverySemantic");
   private static final JsonPointer DELIVERY_SEMANTICS_TYPE_POINTER =
       JsonPointer.compile("/egress/spec/deliverySemantic/type");
+
+  /** @deprecated see {@link #DELIVERY_EXACTLY_ONCE_DURATION_TXN_TIMEOUT_POINTER}. */
   private static final JsonPointer DELIVERY_EXACTLY_ONCE_TXN_TIMEOUT_POINTER =
       JsonPointer.compile("/egress/spec/deliverySemantic/transactionTimeoutMillis");
 
+  private static final JsonPointer DELIVERY_EXACTLY_ONCE_DURATION_TXN_TIMEOUT_POINTER =
+      JsonPointer.compile("/egress/spec/deliverySemantic/transactionTimeout");
+
   static String kafkaAddress(JsonNode json) {
     return Selectors.textAt(json, ADDRESS_POINTER);
   }
@@ -77,7 +83,14 @@
   }
 
   static Duration exactlyOnceDeliveryTxnTimeout(JsonNode json) {
-    long transactionTimeout = Selectors.longAt(json, DELIVERY_EXACTLY_ONCE_TXN_TIMEOUT_POINTER);
-    return Duration.ofMillis(transactionTimeout);
+    // Prefer deprecated millis based timeout for backwards compatibility
+    // then fallback to duration based configuration.
+    OptionalLong transactionTimeoutMilli =
+        Selectors.optionalLongAt(json, DELIVERY_EXACTLY_ONCE_TXN_TIMEOUT_POINTER);
+    if (transactionTimeoutMilli.isPresent()) {
+      return Duration.ofMillis(transactionTimeoutMilli.getAsLong());
+    }
+
+    return Selectors.durationAt(json, DELIVERY_EXACTLY_ONCE_DURATION_TXN_TIMEOUT_POINTER);
   }
 }
diff --git a/statefun-flink/statefun-flink-io-bundle/src/test/resources/generic-kafka-egress.yaml b/statefun-flink/statefun-flink-io-bundle/src/test/resources/generic-kafka-egress.yaml
index ddc9d1e..2014cb3 100644
--- a/statefun-flink/statefun-flink-io-bundle/src/test/resources/generic-kafka-egress.yaml
+++ b/statefun-flink/statefun-flink-io-bundle/src/test/resources/generic-kafka-egress.yaml
@@ -21,6 +21,6 @@
     address: kafka-broker:9092
     deliverySemantic:
       type: exactly-once
-      transactionTimeoutMillis: 100000
+      transactionTimeout: 15min
     properties:
       - foo.config: bar