diff --git a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/PositionRestartIntegrationTest.java b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/PositionRestartIntegrationTest.java
index 66d65f2742338..19edc629dc643 100644
--- a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/PositionRestartIntegrationTest.java
+++ b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/PositionRestartIntegrationTest.java
@@ -110,14 +110,20 @@ public class PositionRestartIntegrationTest {
     private static final long SEED = new Random().nextLong();
     private static final int NUM_BROKERS = 3;
     public static final Duration WINDOW_SIZE = Duration.ofMinutes(5);
-    private static int port = 0;
     private static final String INPUT_TOPIC_NAME = "input-topic";
-    private static final Position INPUT_POSITION = Position.emptyPosition();
     private static final String STORE_NAME = "kv-store";
     private static final long RECORD_TIME = System.currentTimeMillis();
     private static final long WINDOW_START =
         (RECORD_TIME / WINDOW_SIZE.toMillis()) * WINDOW_SIZE.toMillis();
-    public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(NUM_BROKERS);
+
+    // The cluster and the position it produces must be per-instance, not static. Subclasses (e.g. the
+    // transactional variant) inherit this class's @BeforeAll/@AfterAll, so a static cluster would be
+    // shared across both test classes: whichever ran second would find an already-closed
+    // KafkaClusterTestKit, whose executor service cannot be restarted. Under PER_CLASS lifecycle each
+    // test class gets its own instance, and therefore its own cluster.
+    private final EmbeddedKafkaCluster cluster = new EmbeddedKafkaCluster(NUM_BROKERS);
+    private final Position inputPosition = Position.emptyPosition();
+    private int port = 0;
     private KafkaStreams kafkaStreams;
 
     public enum StoresToTest {
@@ -281,16 +287,16 @@ protected Stream<Arguments> data() {
     }
 
     @BeforeAll
-    public static void before()
+    public void before()
         throws InterruptedException, IOException, ExecutionException, TimeoutException {
 
-        CLUSTER.start();
-        CLUSTER.deleteAllTopics();
+        cluster.start();
+        cluster.deleteAllTopics();
         final int partitions = 2;
-        CLUSTER.createTopic(INPUT_TOPIC_NAME, partitions, 1);
+        cluster.createTopic(INPUT_TOPIC_NAME, partitions, 1);
 
         final Properties producerProps = new Properties();
-        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers());
         producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
         producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
 
@@ -315,7 +321,7 @@ public static void before()
             for (final Future<RecordMetadata> future : futures) {
                 final RecordMetadata recordMetadata = future.get(1, TimeUnit.MINUTES);
                 assertThat(recordMetadata.hasOffset(), is(true));
-                INPUT_POSITION.withComponent(
+                inputPosition.withComponent(
                     recordMetadata.topic(),
                     recordMetadata.partition(),
                     recordMetadata.offset()
@@ -323,7 +329,7 @@ public static void before()
             }
         }
 
-        assertThat(INPUT_POSITION, equalTo(
+        assertThat(inputPosition, equalTo(
             Position
                 .emptyPosition()
                 .withComponent(INPUT_TOPIC_NAME, 0, 1L)
@@ -366,8 +372,8 @@ public void afterTest() {
     }
 
     @AfterAll
-    public static void after() {
-        CLUSTER.stop();
+    public void after() {
+        cluster.stop();
     }
 
     @ParameterizedTest
@@ -405,12 +411,12 @@ private void shouldReachExpectedPosition(final Query<?> query) {
             inStore(STORE_NAME)
                 .withQuery(query)
                 .withPartitions(Set.of(0, 1))
-                .withPositionBound(PositionBound.at(INPUT_POSITION));
+                .withPositionBound(PositionBound.at(inputPosition));
 
         final StateQueryResult<?> result =
             IntegrationTestUtils.iqv2WaitForResult(kafkaStreams, request);
 
-        assertThat(result.getPosition(), is(INPUT_POSITION));
+        assertThat(result.getPosition(), is(inputPosition));
     }
 
     private static void setUpSessionDSLTopology(final SessionBytesStoreSupplier supplier,
@@ -668,7 +674,7 @@ protected Properties streamsConfiguration(final boolean cache,
         config.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE);
         config.put(StreamsConfig.APPLICATION_ID_CONFIG, "app-" + safeTestName);
         config.put(StreamsConfig.APPLICATION_SERVER_CONFIG, "localhost:" + (++port));
-        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
+        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers());
         config.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());
         config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());
         config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Integer().getClass());