diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
index 96aa03fb04d72..b272d46a6a548 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
@@ -326,23 +326,6 @@ public KafkaProducer(Properties properties, Serializer<K> keySerializer, Seriali
         this(Utils.propsToMap(properties), keySerializer, valueSerializer);
     }
 
-    /**
-     * Check if partitioner is deprecated and log a warning if it is.
-     */
-    @SuppressWarnings("deprecation")
-    private void warnIfPartitionerDeprecated() {
-        // Using DefaultPartitioner and UniformStickyPartitioner is deprecated, see KIP-794.
-        if (partitioner instanceof org.apache.kafka.clients.producer.internals.DefaultPartitioner) {
-            log.warn("DefaultPartitioner is deprecated.  Please clear " + ProducerConfig.PARTITIONER_CLASS_CONFIG
-                    + " configuration setting to get the default partitioning behavior");
-        }
-        if (partitioner instanceof org.apache.kafka.clients.producer.UniformStickyPartitioner) {
-            log.warn("UniformStickyPartitioner is deprecated.  Please clear " + ProducerConfig.PARTITIONER_CLASS_CONFIG
-                    + " configuration setting and set " + ProducerConfig.PARTITIONER_IGNORE_KEYS_CONFIG
-                    + " to 'true' to get the uniform sticky partitioning behavior");
-        }
-    }
-
     // visible for testing
     @SuppressWarnings({"unchecked", "this-escape"})
     KafkaProducer(ProducerConfig config,
@@ -385,7 +368,6 @@ private void warnIfPartitionerDeprecated() {
                     ProducerConfig.PARTITIONER_CLASS_CONFIG,
                     Partitioner.class,
                     Collections.singletonMap(ProducerConfig.CLIENT_ID_CONFIG, clientId));
-            warnIfPartitionerDeprecated();
             this.partitionerIgnoreKeys = config.getBoolean(ProducerConfig.PARTITIONER_IGNORE_KEYS_CONFIG);
             long retryBackoffMs = config.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG);
             long retryBackoffMaxMs = config.getLong(ProducerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java
index ac7b0a191a74e..564171608568a 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java
@@ -119,31 +119,6 @@ public MockProducer(final Cluster cluster,
         this.mockMetrics = new HashMap<>();
     }
 
-    /**
-     * Create a new mock producer with invented metadata the given autoComplete setting and key\value serializers.
-     *
-     * Equivalent to {@link #MockProducer(Cluster, boolean, Partitioner, Serializer, Serializer) new MockProducer(Cluster.empty(), autoComplete, new DefaultPartitioner(), keySerializer, valueSerializer)}
-     */
-    @SuppressWarnings("deprecation")
-    public MockProducer(final boolean autoComplete,
-                        final Serializer<K> keySerializer,
-                        final Serializer<V> valueSerializer) {
-        this(Cluster.empty(), autoComplete, new org.apache.kafka.clients.producer.internals.DefaultPartitioner(), keySerializer, valueSerializer);
-    }
-
-    /**
-     * Create a new mock producer with invented metadata the given autoComplete setting and key\value serializers.
-     *
-     * Equivalent to {@link #MockProducer(Cluster, boolean, Partitioner, Serializer, Serializer) new MockProducer(cluster, autoComplete, new DefaultPartitioner(), keySerializer, valueSerializer)}
-     */
-    @SuppressWarnings("deprecation")
-    public MockProducer(final Cluster cluster,
-                        final boolean autoComplete,
-                        final Serializer<K> keySerializer,
-                        final Serializer<V> valueSerializer) {
-        this(cluster, autoComplete, new org.apache.kafka.clients.producer.internals.DefaultPartitioner(), keySerializer, valueSerializer);
-    }
-
     /**
      * Create a new mock producer with invented metadata the given autoComplete setting, partitioner and key\value serializers.
      *
@@ -563,6 +538,9 @@ private int partition(ProducerRecord<K, V> record, Cluster cluster) {
         }
         byte[] keyBytes = keySerializer.serialize(topic, record.headers(), record.key());
         byte[] valueBytes = valueSerializer.serialize(topic, record.headers(), record.value());
+        if (partitioner == null) {
+            return this.cluster.partitionsForTopic(record.topic()).get(0).partition();
+        }
         return this.partitioner.partition(topic, record.key(), keyBytes, record.value(), valueBytes, cluster);
     }
 
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/Partitioner.java b/clients/src/main/java/org/apache/kafka/clients/producer/Partitioner.java
index 3db3c3a31eb76..d95bb43d6b80f 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/Partitioner.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/Partitioner.java
@@ -44,7 +44,7 @@ public interface Partitioner extends Configurable, Closeable {
     void close();
 
     /**
-     * Note this method is only implemented in DefaultPartitioner and {@link UniformStickyPartitioner} which
+     * Note this method is only implemented in DefaultPartitioner and UniformStickyPartitioner which
      * are now deprecated. See <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-794%3A+Strictly+Uniform+Sticky+Partitioner">KIP-794</a> for more info.
      * <p>
      * Notifies the partitioner a new batch is about to be created. When using the sticky partitioner,
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/UniformStickyPartitioner.java b/clients/src/main/java/org/apache/kafka/clients/producer/UniformStickyPartitioner.java
deleted file mode 100644
index d9faa5f162427..0000000000000
--- a/clients/src/main/java/org/apache/kafka/clients/producer/UniformStickyPartitioner.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.kafka.clients.producer;
-
-import org.apache.kafka.clients.producer.internals.StickyPartitionCache;
-import org.apache.kafka.common.Cluster;
-
-import java.util.Map;
-
-
-/**
- * @deprecated Since 3.3.0, in order to use default partitioning logic
- * remove the {@code partitioner.class} configuration setting and set {@code partitioner.ignore.keys=true}.
- * See <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-794%3A+Strictly+Uniform+Sticky+Partitioner">KIP-794</a> for more info.
- *
- * The partitioning strategy:
- * <ul>
- * <li>If a partition is specified in the record, use it
- * <li>Otherwise choose the sticky partition that changes when the batch is full.
- * 
- * NOTE: In contrast to the DefaultPartitioner, the record key is NOT used as part of the partitioning strategy in this
- *       partitioner. Records with the same key are not guaranteed to be sent to the same partition.
- * 
- * See <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-480%3A+Sticky+Partitioner">KIP-480</a> for details about sticky partitioning.
- */
-@Deprecated
-public class UniformStickyPartitioner implements Partitioner {
-
-    private final StickyPartitionCache stickyPartitionCache = new StickyPartitionCache();
-
-    public void configure(Map<String, ?> configs) {}
-
-    /**
-     * Compute the partition for the given record.
-     *
-     * @param topic The topic name
-     * @param key The key to partition on (or null if no key)
-     * @param keyBytes serialized key to partition on (or null if no key)
-     * @param value The value to partition on or null
-     * @param valueBytes serialized value to partition on or null
-     * @param cluster The current cluster metadata
-     */
-    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
-        return stickyPartitionCache.partition(topic, cluster);
-    }
-
-    public void close() {}
-    
-    /**
-     * If a batch completed for the current sticky partition, change the sticky partition. 
-     * Alternately, if no sticky partition has been determined, set one.
-     */
-    @SuppressWarnings("deprecation")
-    public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
-        stickyPartitionCache.nextPartition(topic, cluster, prevPartition);
-    }
-}
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/DefaultPartitioner.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/DefaultPartitioner.java
deleted file mode 100644
index 716773626c8c1..0000000000000
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/DefaultPartitioner.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.kafka.clients.producer.internals;
-
-import org.apache.kafka.clients.producer.Partitioner;
-import org.apache.kafka.common.Cluster;
-
-import java.util.Map;
-
-/**
- * NOTE this partitioner is deprecated and shouldn't be used.  To use default partitioning logic
- * remove partitioner.class configuration setting.  See KIP-794 for more info.
- *
- * The default partitioning strategy:
- * <ul>
- * <li>If a partition is specified in the record, use it
- * <li>If no partition is specified but a key is present choose a partition based on a hash of the key
- * <li>If no partition or key is present choose the sticky partition that changes when the batch is full.
- * 
- * See KIP-480 for details about sticky partitioning.
- */
-@Deprecated
-public class DefaultPartitioner implements Partitioner {
-
-    private final StickyPartitionCache stickyPartitionCache = new StickyPartitionCache();
-
-    public void configure(Map<String, ?> configs) {}
-
-    /**
-     * Compute the partition for the given record.
-     *
-     * @param topic The topic name
-     * @param key The key to partition on (or null if no key)
-     * @param keyBytes serialized key to partition on (or null if no key)
-     * @param value The value to partition on or null
-     * @param valueBytes serialized value to partition on or null
-     * @param cluster The current cluster metadata
-     */
-    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
-        return partition(topic, key, keyBytes, value, valueBytes, cluster, cluster.partitionsForTopic(topic).size());
-    }
-
-    /**
-     * Compute the partition for the given record.
-     *
-     * @param topic The topic name
-     * @param numPartitions The number of partitions of the given {@code topic}
-     * @param key The key to partition on (or null if no key)
-     * @param keyBytes serialized key to partition on (or null if no key)
-     * @param value The value to partition on or null
-     * @param valueBytes serialized value to partition on or null
-     * @param cluster The current cluster metadata
-     */
-    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster,
-                         int numPartitions) {
-        if (keyBytes == null) {
-            return stickyPartitionCache.partition(topic, cluster);
-        }
-        return BuiltInPartitioner.partitionForKey(keyBytes, numPartitions);
-    }
-
-    public void close() {}
-    
-    /**
-     * If a batch completed for the current sticky partition, change the sticky partition. 
-     * Alternately, if no sticky partition has been determined, set one.
-     */
-    @SuppressWarnings("deprecation")
-    public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
-        stickyPartitionCache.nextPartition(topic, cluster, prevPartition);
-    }
-}
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/StickyPartitionCache.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/StickyPartitionCache.java
deleted file mode 100644
index 1a00473f06e8a..0000000000000
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/StickyPartitionCache.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.kafka.clients.producer.internals;
-
-import org.apache.kafka.common.Cluster;
-import org.apache.kafka.common.PartitionInfo;
-import org.apache.kafka.common.utils.Utils;
-
-import java.util.List;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ThreadLocalRandom;
-
-/**
- * An internal class that implements a cache used for sticky partitioning behavior. The cache tracks the current sticky
- * partition for any given topic. This class should not be used externally. 
- */
-public class StickyPartitionCache {
-    private final ConcurrentMap<String, Integer> indexCache;
-    public StickyPartitionCache() {
-        this.indexCache = new ConcurrentHashMap<>();
-    }
-
-    public int partition(String topic, Cluster cluster) {
-        Integer part = indexCache.get(topic);
-        if (part == null) {
-            return nextPartition(topic, cluster, -1);
-        }
-        return part;
-    }
-
-    public int nextPartition(String topic, Cluster cluster, int prevPartition) {
-        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
-        Integer oldPart = indexCache.get(topic);
-        Integer newPart = oldPart;
-        // Check that the current sticky partition for the topic is either not set or that the partition that 
-        // triggered the new batch matches the sticky partition that needs to be changed.
-        if (oldPart == null || oldPart == prevPartition) {
-            List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
-            if (availablePartitions.isEmpty()) {
-                int random = Utils.toPositive(ThreadLocalRandom.current().nextInt());
-                newPart = random % partitions.size();
-            } else if (availablePartitions.size() == 1) {
-                newPart = availablePartitions.get(0).partition();
-            } else {
-                while (newPart == null || newPart.equals(oldPart)) {
-                    int random = Utils.toPositive(ThreadLocalRandom.current().nextInt());
-                    newPart = availablePartitions.get(random % availablePartitions.size()).partition();
-                }
-            }
-            // Only change the sticky partition if it is null or prevPartition matches the current sticky partition.
-            if (oldPart == null) {
-                indexCache.putIfAbsent(topic, newPart);
-            } else {
-                indexCache.replace(topic, prevPartition, newPart);
-            }
-            return indexCache.get(topic);
-        }
-        return indexCache.get(topic);
-    }
-
-}
\ No newline at end of file
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/MockProducerTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/MockProducerTest.java
index d3964fc87ed7f..6ec8164c26805 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/MockProducerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/MockProducerTest.java
@@ -59,7 +59,7 @@ public class MockProducerTest {
     private final String groupId = "group";
 
     private void buildMockProducer(boolean autoComplete) {
-        this.producer = new MockProducer<>(autoComplete, new MockSerializer(), new MockSerializer());
+        this.producer = new MockProducer<>(Cluster.empty(), autoComplete, null, new MockSerializer(), new MockSerializer());
     }
 
     @AfterEach
@@ -87,10 +87,16 @@ public void testPartitioner() throws Exception {
         PartitionInfo partitionInfo1 = new PartitionInfo(topic, 1, null, null, null);
         Cluster cluster = new Cluster(null, new ArrayList<>(0), asList(partitionInfo0, partitionInfo1),
                 Collections.emptySet(), Collections.emptySet());
-        MockProducer<String, String> producer = new MockProducer<>(cluster, true, new StringSerializer(), new StringSerializer());
+        MockProducer<String, String> producer = new MockProducer<>(
+            cluster,
+            true,
+            new org.apache.kafka.clients.producer.RoundRobinPartitioner(),
+            new StringSerializer(),
+            new StringSerializer()
+        );
         ProducerRecord<String, String> record = new ProducerRecord<>(topic, "key", "value");
         Future<RecordMetadata> metadata = producer.send(record);
-        assertEquals(1, metadata.get().partition(), "Partition should be correct");
+        assertEquals(0, metadata.get().partition(), "Partition should be correct");
         producer.clear();
         assertEquals(0, producer.history().size(), "Clear should erase our history");
         producer.close();
@@ -680,7 +686,7 @@ public void shouldNotThrowOnFlushProducerIfProducerIsFenced() {
     @Test
     @SuppressWarnings("unchecked")
     public void shouldThrowClassCastException() {
-        try (MockProducer<Integer, String> customProducer = new MockProducer<>(true, new IntegerSerializer(), new StringSerializer())) {
+        try (MockProducer<Integer, String> customProducer = new MockProducer<>(Cluster.empty(), true, null, new IntegerSerializer(), new StringSerializer())) {
             assertThrows(ClassCastException.class, () -> customProducer.send(new ProducerRecord(topic, "key1", "value1")));
         }
     }
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/UniformStickyPartitionerTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/UniformStickyPartitionerTest.java
deleted file mode 100644
index dbaa9dcdecd84..0000000000000
--- a/clients/src/test/java/org/apache/kafka/clients/producer/UniformStickyPartitionerTest.java
+++ /dev/null
@@ -1,211 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.kafka.clients.producer;
-
-import org.apache.kafka.common.Cluster;
-import org.apache.kafka.common.Node;
-import org.apache.kafka.common.PartitionInfo;
-
-import org.junit.jupiter.api.Test;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import static java.util.Arrays.asList;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-public class UniformStickyPartitionerTest {
-    private static final Node[] NODES = new Node[] {
-        new Node(0, "localhost", 99),
-        new Node(1, "localhost", 100),
-        new Node(2, "localhost", 101)
-    };
-
-    private static final String TOPIC_A = "TOPIC_A";
-    private static final String TOPIC_B = "TOPIC_B";
-
-    @SuppressWarnings("deprecation")
-    @Test
-    public void testRoundRobinWithUnavailablePartitions() {
-        // Intentionally make the partition list not in partition order to test the edge
-        // cases.
-        List<PartitionInfo> partitions = asList(
-                new PartitionInfo("test", 1, null, NODES, NODES),
-                new PartitionInfo("test", 2, NODES[1], NODES, NODES),
-                new PartitionInfo("test", 0, NODES[0], NODES, NODES));
-        // When there are some unavailable partitions, we want to make sure that (1) we
-        // always pick an available partition,
-        // and (2) the available partitions are selected in a sticky way.
-        int countForPart0 = 0;
-        int countForPart2 = 0;
-        int part = 0;
-        Partitioner partitioner = new UniformStickyPartitioner();
-        Cluster cluster = new Cluster("clusterId", asList(NODES[0], NODES[1], NODES[2]), partitions,
-            Collections.emptySet(), Collections.emptySet());
-        for (int i = 0; i < 50; i++) {
-            part = partitioner.partition("test", null, null, null, null, cluster);
-            assertTrue(part == 0 || part == 2, "We should never choose a leader-less node in round robin");
-            if (part == 0)
-                countForPart0++;
-            else
-                countForPart2++;
-        }
-        // Simulates switching the sticky partition on a new batch.
-        partitioner.onNewBatch("test", cluster, part);       
-        for (int i = 1; i <= 50; i++) {
-            part = partitioner.partition("test", null, null, null, null, cluster);
-            assertTrue(part == 0 || part == 2, "We should never choose a leader-less node in round robin");
-            if (part == 0)
-                countForPart0++;
-            else
-                countForPart2++;
-        }
-        assertEquals(countForPart0, countForPart2, "The distribution between two available partitions should be even");
-    }
-
-    @SuppressWarnings("deprecation")
-    @Test
-    public void testRoundRobinWithKeyBytes() {
-        List<PartitionInfo> allPartitions = asList(new PartitionInfo(TOPIC_A, 0, NODES[0], NODES, NODES),
-                new PartitionInfo(TOPIC_A, 1, NODES[1], NODES, NODES), new PartitionInfo(TOPIC_A, 2, NODES[1], NODES, NODES),
-                new PartitionInfo(TOPIC_B, 0, NODES[0], NODES, NODES));
-        Cluster testCluster = new Cluster("clusterId", asList(NODES[0], NODES[1], NODES[2]), allPartitions,
-                Collections.emptySet(), Collections.emptySet());
-
-        final Map<Integer, Integer> partitionCount = new HashMap<>();
-
-        final byte[] keyBytes = "key".getBytes();
-        int partition = 0;
-        Partitioner partitioner = new UniformStickyPartitioner();
-        for (int i = 0; i < 30; ++i) {
-            partition = partitioner.partition(TOPIC_A, null, keyBytes, null, null, testCluster);
-            Integer count = partitionCount.get(partition);
-            if (null == count)
-                count = 0;
-            partitionCount.put(partition, count + 1);
-
-            if (i % 5 == 0) {
-                partitioner.partition(TOPIC_B, null, keyBytes, null, null, testCluster);
-            }
-        }
-        // Simulate a batch filling up and switching the sticky partition.
-        partitioner.onNewBatch(TOPIC_A, testCluster, partition);
-        partitioner.onNewBatch(TOPIC_B, testCluster, 0);
-        
-        // Save old partition to ensure that the wrong partition does not trigger a new batch.
-        int oldPart = partition; 
-        
-        for (int i = 0; i < 30; ++i) {
-            partition = partitioner.partition(TOPIC_A, null, keyBytes, null, null, testCluster);
-            Integer count = partitionCount.get(partition);
-            if (null == count)
-                count = 0;
-            partitionCount.put(partition, count + 1);
-
-            if (i % 5 == 0) {
-                partitioner.partition(TOPIC_B, null, keyBytes, null, null, testCluster);
-            }
-        }
-        
-        int newPart = partition;
-        
-        // Attempt to switch the partition with the wrong previous partition. Sticky partition should not change.
-        partitioner.onNewBatch(TOPIC_A, testCluster, oldPart);
-        
-        for (int i = 0; i < 30; ++i) {
-            partition = partitioner.partition(TOPIC_A, null, keyBytes, null, null, testCluster);
-            Integer count = partitionCount.get(partition);
-            if (null == count)
-                count = 0;
-            partitionCount.put(partition, count + 1);
-
-            if (i % 5 == 0) {
-                partitioner.partition(TOPIC_B, null, keyBytes, null, null, testCluster);
-            }
-        } 
-               
-        assertEquals(30, partitionCount.get(oldPart).intValue());
-        assertEquals(60, partitionCount.get(newPart).intValue());
-    }
-
-    @SuppressWarnings("deprecation")
-    @Test
-    public void testRoundRobinWithNullKeyBytes() {
-        List<PartitionInfo> allPartitions = asList(new PartitionInfo(TOPIC_A, 0, NODES[0], NODES, NODES),
-                new PartitionInfo(TOPIC_A, 1, NODES[1], NODES, NODES), new PartitionInfo(TOPIC_A, 2, NODES[1], NODES, NODES),
-                new PartitionInfo(TOPIC_B, 0, NODES[0], NODES, NODES));
-        Cluster testCluster = new Cluster("clusterId", asList(NODES[0], NODES[1], NODES[2]), allPartitions,
-                Collections.emptySet(), Collections.emptySet());
-
-        final Map<Integer, Integer> partitionCount = new HashMap<>();
-
-        int partition = 0;
-        Partitioner partitioner = new UniformStickyPartitioner();
-        for (int i = 0; i < 30; ++i) {
-            partition = partitioner.partition(TOPIC_A, null, null, null, null, testCluster);
-            Integer count = partitionCount.get(partition);
-            if (null == count)
-                count = 0;
-            partitionCount.put(partition, count + 1);
-
-            if (i % 5 == 0) {
-                partitioner.partition(TOPIC_B, null, null, null, null, testCluster);
-            }
-        }
-        // Simulate a batch filling up and switching the sticky partition.
-        partitioner.onNewBatch(TOPIC_A, testCluster, partition);
-        partitioner.onNewBatch(TOPIC_B, testCluster, 0);
-
-        // Save old partition to ensure that the wrong partition does not trigger a new batch.
-        int oldPart = partition;
-
-        for (int i = 0; i < 30; ++i) {
-            partition = partitioner.partition(TOPIC_A, null, null, null, null, testCluster);
-            Integer count = partitionCount.get(partition);
-            if (null == count)
-                count = 0;
-            partitionCount.put(partition, count + 1);
-
-            if (i % 5 == 0) {
-                partitioner.partition(TOPIC_B, null, null, null, null, testCluster);
-            }
-        }
-
-        int newPart = partition;
-
-        // Attempt to switch the partition with the wrong previous partition. Sticky partition should not change.
-        partitioner.onNewBatch(TOPIC_A, testCluster, oldPart);
-
-        for (int i = 0; i < 30; ++i) {
-            partition = partitioner.partition(TOPIC_A, null, null, null, null, testCluster);
-            Integer count = partitionCount.get(partition);
-            if (null == count)
-                count = 0;
-            partitionCount.put(partition, count + 1);
-
-            if (i % 5 == 0) {
-                partitioner.partition(TOPIC_B, null, null, null, null, testCluster);
-            }
-        }
-
-        assertEquals(30, partitionCount.get(oldPart).intValue());
-        assertEquals(60, partitionCount.get(newPart).intValue());
-    }
-}
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/DefaultPartitionerTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/DefaultPartitionerTest.java
deleted file mode 100644
index 353e07b366af6..0000000000000
--- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/DefaultPartitionerTest.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.kafka.clients.producer.internals;
-
-import org.apache.kafka.clients.producer.Partitioner;
-import org.apache.kafka.common.Cluster;
-import org.apache.kafka.common.Node;
-import org.apache.kafka.common.PartitionInfo;
-
-import org.junit.jupiter.api.Test;
-
-import java.util.Collections;
-import java.util.List;
-
-import static java.util.Arrays.asList;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-public class DefaultPartitionerTest {
-    private static final byte[] KEY_BYTES = "key".getBytes();
-    private static final Node[] NODES = new Node[] {
-        new Node(0, "localhost", 99),
-        new Node(1, "localhost", 100),
-        new Node(12, "localhost", 101)
-    };
-    private static final String TOPIC = "test";
-    // Intentionally make the partition list not in partition order to test the edge cases.
-    private static final List<PartitionInfo> PARTITIONS = asList(new PartitionInfo(TOPIC, 1, null, NODES, NODES),
-                                                    new PartitionInfo(TOPIC, 2, NODES[1], NODES, NODES),
-                                                    new PartitionInfo(TOPIC, 0, NODES[0], NODES, NODES));
-
-    @Test
-    public void testKeyPartitionIsStable() {
-        @SuppressWarnings("deprecation")
-        final Partitioner partitioner = new DefaultPartitioner();
-        final Cluster cluster = new Cluster("clusterId", asList(NODES), PARTITIONS,
-            Collections.emptySet(), Collections.emptySet());
-        int partition = partitioner.partition(TOPIC,  null, KEY_BYTES, null, null, cluster);
-        assertEquals(partition, partitioner.partition(TOPIC, null, KEY_BYTES, null, null, cluster), "Same key should yield same partition");
-    }
-}
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java
index 169b6b868439c..5eefcc4e974ce 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java
@@ -21,7 +21,6 @@
 import org.apache.kafka.clients.MetadataSnapshot;
 import org.apache.kafka.clients.NodeApiVersions;
 import org.apache.kafka.clients.producer.Callback;
-import org.apache.kafka.clients.producer.Partitioner;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.Cluster;
 import org.apache.kafka.common.KafkaException;
@@ -1213,79 +1212,6 @@ public void testExpiredBatchesRetry() throws InterruptedException {
         }
     }
 
-    @SuppressWarnings("deprecation")
-    @Test
-    public void testStickyBatches() throws Exception {
-        long now = time.milliseconds();
-
-        // Test case assumes that the records do not fill the batch completely
-        int batchSize = 1025;
-
-        Partitioner partitioner = new DefaultPartitioner();
-        RecordAccumulator accum = createTestRecordAccumulator(3200,
-            batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10L * batchSize, Compression.NONE, 10);
-        int expectedAppends = expectedNumAppendsNoKey(batchSize);
-
-        // Create first batch
-        int partition = partitioner.partition(topic, null, null, "value", value, cluster);
-        accum.append(topic, partition, 0L, null, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs, false, time.milliseconds(), cluster);
-        int appends = 1;
-
-        boolean switchPartition = false;
-        while (!switchPartition) {
-            // Append to the first batch
-            partition = partitioner.partition(topic, null, null, "value", value, cluster);
-            RecordAccumulator.RecordAppendResult result = accum.append(topic, partition, 0L, null,
-                value, Record.EMPTY_HEADERS, null, maxBlockTimeMs, true, time.milliseconds(), cluster);
-            Deque<ProducerBatch> partitionBatches1 = accum.getDeque(tp1);
-            Deque<ProducerBatch> partitionBatches2 = accum.getDeque(tp2);
-            Deque<ProducerBatch> partitionBatches3 = accum.getDeque(tp3);
-            int numBatches = (partitionBatches1 == null ? 0 : partitionBatches1.size()) + (partitionBatches2 == null ? 0 : partitionBatches2.size()) + (partitionBatches3 == null ? 0 : partitionBatches3.size());
-            // Only one batch is created because the partition is sticky.
-            assertEquals(1, numBatches);
-
-            switchPartition = result.abortForNewBatch;
-            // We only appended if we do not retry.
-            if (!switchPartition) {
-                appends++;
-                assertEquals(0, accum.ready(metadataCache, now).readyNodes.size(), "No partitions should be ready.");
-            }
-        }
-
-        // Batch should be full.
-        assertEquals(1, accum.ready(metadataCache, time.milliseconds()).readyNodes.size());
-        assertEquals(appends, expectedAppends);
-        switchPartition = false;
-
-        // KafkaProducer would call this method in this case, make second batch
-        partitioner.onNewBatch(topic, cluster, partition);
-        partition = partitioner.partition(topic, null, null, "value", value, cluster);
-        accum.append(topic, partition, 0L, null, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs, false, time.milliseconds(), cluster);
-        appends++;
-
-        // These appends all go into the second batch
-        while (!switchPartition) {
-            partition = partitioner.partition(topic, null, null, "value", value, cluster);
-            RecordAccumulator.RecordAppendResult result = accum.append(topic, partition, 0L, null, value,
-                Record.EMPTY_HEADERS, null, maxBlockTimeMs, true, time.milliseconds(), cluster);
-            Deque<ProducerBatch> partitionBatches1 = accum.getDeque(tp1);
-            Deque<ProducerBatch> partitionBatches2 = accum.getDeque(tp2);
-            Deque<ProducerBatch> partitionBatches3 = accum.getDeque(tp3);
-            int numBatches = (partitionBatches1 == null ? 0 : partitionBatches1.size()) + (partitionBatches2 == null ? 0 : partitionBatches2.size()) + (partitionBatches3 == null ? 0 : partitionBatches3.size());
-            // Only two batches because the new partition is also sticky.
-            assertEquals(2, numBatches);
-
-            switchPartition = result.abortForNewBatch;
-            // We only appended if we do not retry.
-            if (!switchPartition) {
-                appends++;
-            }
-        }
-
-        // There should be two full batches now.
-        assertEquals(appends, 2 * expectedAppends);
-    }
-
     @Test
     public void testUniformBuiltInPartitioner() throws Exception {
 
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/StickyPartitionCacheTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/StickyPartitionCacheTest.java
deleted file mode 100644
index 8f8f1c06bd988..0000000000000
--- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/StickyPartitionCacheTest.java
+++ /dev/null
@@ -1,112 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.kafka.clients.producer.internals;
-
-import org.apache.kafka.common.Cluster;
-import org.apache.kafka.common.Node;
-import org.apache.kafka.common.PartitionInfo;
-
-import org.junit.jupiter.api.Test;
-
-import java.util.Collections;
-import java.util.List;
-
-import static java.util.Arrays.asList;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotEquals;
-
-public class StickyPartitionCacheTest {
-    private static final Node[] NODES = new Node[] {
-        new Node(0, "localhost", 99),
-        new Node(1, "localhost", 100),
-        new Node(2, "localhost", 101),
-        new Node(11, "localhost", 102)
-    };
-    static final String TOPIC_A = "topicA";
-    static final String TOPIC_B = "topicB";
-    static final String TOPIC_C = "topicC";
-
-    @Test
-    public void testStickyPartitionCache() {
-        List<PartitionInfo> allPartitions = asList(new PartitionInfo(TOPIC_A, 0, NODES[0], NODES, NODES),
-            new PartitionInfo(TOPIC_A, 1, NODES[1], NODES, NODES),
-            new PartitionInfo(TOPIC_A, 2, NODES[2], NODES, NODES),
-            new PartitionInfo(TOPIC_B, 0, NODES[0], NODES, NODES)
-        );
-        Cluster testCluster = new Cluster("clusterId", asList(NODES), allPartitions,
-            Collections.emptySet(), Collections.emptySet());
-        StickyPartitionCache stickyPartitionCache = new StickyPartitionCache();
-
-        int partA = stickyPartitionCache.partition(TOPIC_A, testCluster);
-        assertEquals(partA, stickyPartitionCache.partition(TOPIC_A, testCluster));
-
-        int partB = stickyPartitionCache.partition(TOPIC_B, testCluster);
-        assertEquals(partB, stickyPartitionCache.partition(TOPIC_B, testCluster));
-
-        int changedPartA = stickyPartitionCache.nextPartition(TOPIC_A, testCluster, partA);
-        assertEquals(changedPartA, stickyPartitionCache.partition(TOPIC_A, testCluster));
-        assertNotEquals(partA, changedPartA);
-        int changedPartA2 = stickyPartitionCache.partition(TOPIC_A, testCluster);
-        assertEquals(changedPartA2, changedPartA);
-
-        // We do not want to change partitions because the previous partition does not match the current sticky one.
-        int changedPartA3 = stickyPartitionCache.nextPartition(TOPIC_A, testCluster, partA);
-        assertEquals(changedPartA3, changedPartA2);
-
-        // Check that we can still use the partitioner when there is only one partition
-        int changedPartB = stickyPartitionCache.nextPartition(TOPIC_B, testCluster, partB);
-        assertEquals(changedPartB, stickyPartitionCache.partition(TOPIC_B, testCluster));
-    }
-    
-    @Test
-    public void unavailablePartitionsTest() {
-        // Partition 1 in topic A and partition 0 in topic B are unavailable partitions.
-        List<PartitionInfo> allPartitions = asList(new PartitionInfo(TOPIC_A, 0, NODES[0], NODES, NODES),
-            new PartitionInfo(TOPIC_A, 1, null, NODES, NODES),
-            new PartitionInfo(TOPIC_A, 2, NODES[2], NODES, NODES),
-            new PartitionInfo(TOPIC_B, 0, null, NODES, NODES),
-            new PartitionInfo(TOPIC_B, 1, NODES[0], NODES, NODES),
-            new PartitionInfo(TOPIC_C, 0, null, NODES, NODES)
-        );
-        
-        Cluster testCluster = new Cluster("clusterId", asList(NODES[0], NODES[1], NODES[2]), allPartitions,
-            Collections.emptySet(), Collections.emptySet());
-        StickyPartitionCache stickyPartitionCache = new StickyPartitionCache();
-        
-        // Assure we never choose partition 1 because it is unavailable.
-        int partA = stickyPartitionCache.partition(TOPIC_A, testCluster);
-        assertNotEquals(1, partA);
-        for (int aPartitions = 0; aPartitions < 100; aPartitions++) {
-            partA = stickyPartitionCache.nextPartition(TOPIC_A, testCluster, partA);
-            assertNotEquals(1, stickyPartitionCache.partition(TOPIC_A, testCluster));
-        }
-        
-        // Assure we always choose partition 1 for topic B.
-        int partB = stickyPartitionCache.partition(TOPIC_B, testCluster);
-        assertEquals(1, partB);
-        for (int bPartitions = 0; bPartitions < 100; bPartitions++) {
-            partB = stickyPartitionCache.nextPartition(TOPIC_B, testCluster, partB);
-            assertEquals(1, stickyPartitionCache.partition(TOPIC_B, testCluster));
-        }
-        
-        // Assure that we still choose the partition when there are no partitions available.
-        int partC = stickyPartitionCache.partition(TOPIC_C, testCluster);
-        assertEquals(0, partC);
-        partC = stickyPartitionCache.nextPartition(TOPIC_C, testCluster, partC);
-        assertEquals(0, partC);
-    }
-}
diff --git a/docs/upgrade.html b/docs/upgrade.html
index 4380fbd931dbc..0d943f7765fa7 100644
--- a/docs/upgrade.html
+++ b/docs/upgrade.html
@@ -53,6 +53,8 @@ <h5><a id="upgrade_400_notable" href="#upgrade_400_notable">Notable changes in 4
                         <li>The <code>org.apache.kafka.common.errors.NotLeaderForPartitionException</code> class was removed. The <code>org.apache.kafka.common.errors.NotLeaderOrFollowerException</code> is returned
                             if a request could not be processed because the broker is not the leader or follower for a topic partition.
                         </li>
+                        <li>The <code>org.apache.kafka.clients.producer.internals.DefaultPartitioner</code> and <code>org.apache.kafka.clients.producer.UniformStickyPartitioner</code> class was removed.
+                        </li>
                     </ul>
                 </li>
                 <li><b>Broker</b>
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/WindowedStreamPartitionerTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/WindowedStreamPartitionerTest.java
index eb8f95f96f19b..25a15efdd4c9d 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/WindowedStreamPartitionerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/WindowedStreamPartitionerTest.java
@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.streams.kstream.internals;
 
+import org.apache.kafka.clients.producer.internals.BuiltInPartitioner;
 import org.apache.kafka.common.Cluster;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.PartitionInfo;
@@ -58,8 +59,6 @@ public class WindowedStreamPartitionerTest {
     @Test
     public void testCopartitioning() {
         final Random rand = new Random();
-        @SuppressWarnings("deprecation")
-        final org.apache.kafka.clients.producer.internals.DefaultPartitioner defaultPartitioner = new org.apache.kafka.clients.producer.internals.DefaultPartitioner();
         final WindowedSerializer<Integer> timeWindowedSerializer = new TimeWindowedSerializer<>(intSerializer);
         final WindowedStreamPartitioner<Integer, String> streamPartitioner = new WindowedStreamPartitioner<>(timeWindowedSerializer);
 
@@ -68,9 +67,8 @@ public void testCopartitioning() {
             final byte[] keyBytes = intSerializer.serialize(topicName, key);
 
             final String value = key.toString();
-            final byte[] valueBytes = stringSerializer.serialize(topicName, value);
 
-            final Set<Integer> expected = Collections.singleton(defaultPartitioner.partition(topicName, key, keyBytes, value, valueBytes, cluster));
+            final Set<Integer> expected = Set.of(BuiltInPartitioner.partitionForKey(keyBytes, cluster.partitionsForTopic(topicName).size()));
 
             for (int w = 1; w < 10; w++) {
                 final TimeWindow window = new TimeWindow(10 * w, 20 * w);
@@ -82,7 +80,5 @@ public void testCopartitioning() {
                 assertEquals(expected, actual.get());
             }
         }
-
-        defaultPartitioner.close();
     }
 }
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java
index 66a581fb8f41a..e68d5a4396f78 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java
@@ -136,7 +136,7 @@ public class RecordCollectorTest {
         (topic, key, value, numPartitions) -> Optional.of(Collections.singleton(Integer.parseInt(key) % numPartitions));
 
     private final MockProducer<byte[], byte[]> mockProducer
-        = new MockProducer<>(cluster, true, new ByteArraySerializer(), new ByteArraySerializer());
+        = new MockProducer<>(cluster, true, new org.apache.kafka.clients.producer.RoundRobinPartitioner(), new ByteArraySerializer(), new ByteArraySerializer());
 
     private StreamsProducer streamsProducer;
     private ProcessorTopology topology;
@@ -547,10 +547,10 @@ public Optional<Set<Integer>> partitions(final String topic, final String key, f
 
         final Map<TopicPartition, Long> offsets = collector.offsets();
 
-        // with mock producer without specific partition, we would use default producer partitioner with murmur hash
-        assertEquals(3L, (long) offsets.get(new TopicPartition(topic, 0)));
+        // with mock producer without specific partition, we would use roundrobin producer partitioner
+        assertEquals(2L, (long) offsets.get(new TopicPartition(topic, 0)));
         assertEquals(2L, (long) offsets.get(new TopicPartition(topic, 1)));
-        assertEquals(1L, (long) offsets.get(new TopicPartition(topic, 2)));
+        assertEquals(2L, (long) offsets.get(new TopicPartition(topic, 2)));
         assertEquals(9, mockProducer.history().size());
     }
 
@@ -601,10 +601,10 @@ public void shouldUseDefaultPartitionerAsPartitionReturnsEmptyOptional() {
 
         final Map<TopicPartition, Long> offsets = collector.offsets();
 
-        // with mock producer without specific partition, we would use default producer partitioner with murmur hash
-        assertEquals(3L, (long) offsets.get(new TopicPartition(topic, 0)));
+        // with mock producer without specific partition, we would use roundrobin producer partitioner
+        assertEquals(2L, (long) offsets.get(new TopicPartition(topic, 0)));
         assertEquals(2L, (long) offsets.get(new TopicPartition(topic, 1)));
-        assertEquals(1L, (long) offsets.get(new TopicPartition(topic, 2)));
+        assertEquals(2L, (long) offsets.get(new TopicPartition(topic, 2)));
         assertEquals(9, mockProducer.history().size());
     }
 
@@ -652,10 +652,10 @@ public void shouldUseDefaultPartitionerAsStreamPartitionerIsNull() {
 
         final Map<TopicPartition, Long> offsets = collector.offsets();
 
-        // with mock producer without specific partition, we would use default producer partitioner with murmur hash
-        assertEquals(3L, (long) offsets.get(new TopicPartition(topic, 0)));
+        // with mock producer without specific partition, we would use roundrobin producer partitioner
+        assertEquals(2L, (long) offsets.get(new TopicPartition(topic, 0)));
         assertEquals(2L, (long) offsets.get(new TopicPartition(topic, 1)));
-        assertEquals(1L, (long) offsets.get(new TopicPartition(topic, 2)));
+        assertEquals(2L, (long) offsets.get(new TopicPartition(topic, 2)));
         assertEquals(9, mockProducer.history().size());
     }
 
@@ -675,10 +675,10 @@ public void shouldSendWithNoPartition() {
 
         final Map<TopicPartition, Long> offsets = collector.offsets();
 
-        // with mock producer without specific partition, we would use default producer partitioner with murmur hash
-        assertEquals(3L, (long) offsets.get(new TopicPartition(topic, 0)));
+        // with mock producer without specific partition, we would use roundrobin producer partitioner
+        assertEquals(2L, (long) offsets.get(new TopicPartition(topic, 0)));
         assertEquals(2L, (long) offsets.get(new TopicPartition(topic, 1)));
-        assertEquals(1L, (long) offsets.get(new TopicPartition(topic, 2)));
+        assertEquals(2L, (long) offsets.get(new TopicPartition(topic, 2)));
         assertEquals(9, mockProducer.history().size());
     }
 
@@ -1431,7 +1431,7 @@ public void shouldNotAbortTxnOnEOSCloseDirtyIfNothingSent() {
             logContext,
             taskId,
             new StreamsProducer(
-                new MockProducer<>(cluster, true, byteArraySerializer, byteArraySerializer) {
+                new MockProducer<>(cluster, true, null, byteArraySerializer, byteArraySerializer) {
                     @Override
                     public void abortTransaction() {
                         functionCalled.set(true);
@@ -1456,7 +1456,7 @@ public void shouldThrowIfTopicIsUnknownOnSendWithPartitioner() {
             logContext,
             taskId,
             new StreamsProducer(
-                new MockProducer<>(cluster, true, byteArraySerializer, byteArraySerializer) {
+                new MockProducer<>(cluster, true, null, byteArraySerializer, byteArraySerializer) {
                     @Override
                     public List<PartitionInfo> partitionsFor(final String topic) {
                         return Collections.emptyList();
@@ -1885,7 +1885,7 @@ public byte[] serialize(final String topic, final Headers headers, final String
 
     private StreamsProducer getExceptionalStreamsProducerOnSend(final Exception exception) {
         return new StreamsProducer(
-            new MockProducer<>(cluster, true, byteArraySerializer, byteArraySerializer) {
+            new MockProducer<>(cluster, true, null, byteArraySerializer, byteArraySerializer) {
                 @Override
                 public synchronized Future<RecordMetadata> send(final ProducerRecord<byte[], byte[]> record, final Callback callback) {
                     callback.onCompletion(null, exception);
@@ -1900,7 +1900,7 @@ public synchronized Future<RecordMetadata> send(final ProducerRecord<byte[], byt
 
     private StreamsProducer getExceptionalStreamProducerOnPartitionsFor(final RuntimeException exception) {
         return new StreamsProducer(
-            new MockProducer<>(cluster, true, byteArraySerializer, byteArraySerializer) {
+            new MockProducer<>(cluster, true, null, byteArraySerializer, byteArraySerializer) {
                 @Override
                 public synchronized List<PartitionInfo> partitionsFor(final String topic) {
                     throw exception;
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsProducerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsProducerTest.java
index 3fbf3b912da8f..6c6362ee8bcbc 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsProducerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsProducerTest.java
@@ -110,9 +110,9 @@ public class StreamsProducerTest {
     );
 
     private final MockProducer<byte[], byte[]> nonEosMockProducer
-        = new MockProducer<>(cluster, true, new ByteArraySerializer(), new ByteArraySerializer());
+        = new MockProducer<>(cluster, true, new org.apache.kafka.clients.producer.RoundRobinPartitioner(), new ByteArraySerializer(), new ByteArraySerializer());
     private final MockProducer<byte[], byte[]> eosMockProducer
-        = new MockProducer<>(cluster, true, new ByteArraySerializer(), new ByteArraySerializer());
+        = new MockProducer<>(cluster, true, new org.apache.kafka.clients.producer.RoundRobinPartitioner(), new ByteArraySerializer(), new ByteArraySerializer());
 
     private StreamsProducer nonEosStreamsProducer;
     private StreamsProducer eosStreamsProducer;
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java b/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
index fdfed28ece2d2..78c6dedcbf45c 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
@@ -215,7 +215,7 @@ private KeyValueStoreTestDriver(final StateSerdes<K, V> serdes) {
             logContext,
             new TaskId(0, 0),
             new StreamsProducer(
-                new MockProducer<>(null, true, null, null),
+                new MockProducer<>(null, true, null, null, null),
                 AT_LEAST_ONCE,
                 Time.SYSTEM,
                 logContext
diff --git a/streams/src/test/java/org/apache/kafka/test/MockClientSupplier.java b/streams/src/test/java/org/apache/kafka/test/MockClientSupplier.java
index 37723f1e7cc47..17eb34d803b3e 100644
--- a/streams/src/test/java/org/apache/kafka/test/MockClientSupplier.java
+++ b/streams/src/test/java/org/apache/kafka/test/MockClientSupplier.java
@@ -76,7 +76,7 @@ public Producer<byte[], byte[]> getProducer(final Map<String, Object> config) {
 
         final MockProducer<byte[], byte[]> producer;
         if (preparedProducers.isEmpty()) {
-            producer = new MockProducer<>(cluster, true, BYTE_ARRAY_SERIALIZER, BYTE_ARRAY_SERIALIZER);
+            producer = new MockProducer<>(cluster, true, null, BYTE_ARRAY_SERIALIZER, BYTE_ARRAY_SERIALIZER);
         } else {
             producer = preparedProducers.remove(0);
         }
diff --git a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
index 68b8e3b2a733e..f5eb62b36c30c 100644
--- a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
+++ b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
@@ -25,6 +25,7 @@
 import org.apache.kafka.clients.producer.MockProducer;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.Cluster;
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.PartitionInfo;
@@ -337,7 +338,7 @@ private TopologyTestDriver(final InternalTopologyBuilder builder,
 
         consumer = new MockConsumer<>(AutoOffsetResetStrategy.EARLIEST.name());
         final Serializer<byte[]> bytesSerializer = new ByteArraySerializer();
-        producer = new MockProducer<>(true, bytesSerializer, bytesSerializer) {
+        producer = new MockProducer<>(Cluster.empty(), true, null, bytesSerializer, bytesSerializer) {
             @Override
             public List<PartitionInfo> partitionsFor(final String topic) {
                 return Collections.singletonList(new PartitionInfo(topic, PARTITION_ID, null, null, null));