diff --git a/build.gradle b/build.gradle
index 5051490d333..349345a0338 100644
--- a/build.gradle
+++ b/build.gradle
@@ -1898,7 +1898,7 @@ project(':streams:test-utils') {
 
     testImplementation project(':clients').sourceSets.test.output
     testImplementation libs.junitJupiter
-    testImplementation libs.easymock
+    testImplementation libs.mockitoCore
     testImplementation libs.hamcrest
 
     testRuntimeOnly libs.slf4jlog4j
diff --git a/streams/test-utils/src/test/java/org/apache/kafka/streams/KeyValueStoreFacadeTest.java b/streams/test-utils/src/test/java/org/apache/kafka/streams/KeyValueStoreFacadeTest.java
index 4b45477f1cf..b79a2de9930 100644
--- a/streams/test-utils/src/test/java/org/apache/kafka/streams/KeyValueStoreFacadeTest.java
+++ b/streams/test-utils/src/test/java/org/apache/kafka/streams/KeyValueStoreFacadeTest.java
@@ -17,28 +17,28 @@
 package org.apache.kafka.streams;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.streams.TopologyTestDriver.KeyValueStoreFacade;
 import org.apache.kafka.streams.processor.ProcessorContext;
 import org.apache.kafka.streams.processor.StateStore;
 import org.apache.kafka.streams.processor.StateStoreContext;
 import org.apache.kafka.streams.state.TimestampedKeyValueStore;
 import org.apache.kafka.streams.state.ValueAndTimestamp;
-import org.apache.kafka.streams.TopologyTestDriver.KeyValueStoreFacade;
-import org.easymock.EasyMock;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import static java.util.Arrays.asList;
-import static org.easymock.EasyMock.expect;
-import static org.easymock.EasyMock.expectLastCall;
-import static org.easymock.EasyMock.mock;
-import static org.easymock.EasyMock.replay;
-import static org.easymock.EasyMock.verify;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 public class KeyValueStoreFacadeTest {
-    private final TimestampedKeyValueStore<String, String> mockedKeyValueTimestampStore = EasyMock.mock(TimestampedKeyValueStore.class);
+    @SuppressWarnings("unchecked")
+    private final TimestampedKeyValueStore<String, String> mockedKeyValueTimestampStore = mock(TimestampedKeyValueStore.class);
 
     private KeyValueStoreFacade<String, String> keyValueStoreFacade;
 
@@ -52,124 +52,98 @@ public class KeyValueStoreFacadeTest {
     public void shouldForwardDeprecatedInit() {
         final ProcessorContext context = mock(ProcessorContext.class);
         final StateStore store = mock(StateStore.class);
-        mockedKeyValueTimestampStore.init(context, store);
-        expectLastCall();
-        replay(mockedKeyValueTimestampStore);
 
         keyValueStoreFacade.init(context, store);
-        verify(mockedKeyValueTimestampStore);
+        verify(mockedKeyValueTimestampStore).init(context, store);
     }
 
     @Test
     public void shouldForwardInit() {
         final StateStoreContext context = mock(StateStoreContext.class);
         final StateStore store = mock(StateStore.class);
-        mockedKeyValueTimestampStore.init(context, store);
-        expectLastCall();
-        replay(mockedKeyValueTimestampStore);
 
         keyValueStoreFacade.init(context, store);
-        verify(mockedKeyValueTimestampStore);
+        verify(mockedKeyValueTimestampStore).init(context, store);
     }
 
     @Test
     public void shouldPutWithUnknownTimestamp() {
-        mockedKeyValueTimestampStore.put("key", ValueAndTimestamp.make("value", ConsumerRecord.NO_TIMESTAMP));
-        expectLastCall();
-        replay(mockedKeyValueTimestampStore);
-
         keyValueStoreFacade.put("key", "value");
-        verify(mockedKeyValueTimestampStore);
+        verify(mockedKeyValueTimestampStore)
+            .put("key", ValueAndTimestamp.make("value", ConsumerRecord.NO_TIMESTAMP));
     }
 
     @Test
     public void shouldPutIfAbsentWithUnknownTimestamp() {
-        expect(mockedKeyValueTimestampStore.putIfAbsent("key", ValueAndTimestamp.make("value", ConsumerRecord.NO_TIMESTAMP)))
-            .andReturn(null)
-            .andReturn(ValueAndTimestamp.make("oldValue", 42L));
-        replay(mockedKeyValueTimestampStore);
+        doReturn(null, ValueAndTimestamp.make("oldValue", 42L))
+            .when(mockedKeyValueTimestampStore)
+            .putIfAbsent("key", ValueAndTimestamp.make("value", ConsumerRecord.NO_TIMESTAMP));
 
         assertNull(keyValueStoreFacade.putIfAbsent("key", "value"));
         assertThat(keyValueStoreFacade.putIfAbsent("key", "value"), is("oldValue"));
-        verify(mockedKeyValueTimestampStore);
+        verify(mockedKeyValueTimestampStore, times(2))
+            .putIfAbsent("key", ValueAndTimestamp.make("value", ConsumerRecord.NO_TIMESTAMP));
     }
 
     @Test
     public void shouldPutAllWithUnknownTimestamp() {
-        mockedKeyValueTimestampStore.put("key1", ValueAndTimestamp.make("value1", ConsumerRecord.NO_TIMESTAMP));
-        mockedKeyValueTimestampStore.put("key2", ValueAndTimestamp.make("value2", ConsumerRecord.NO_TIMESTAMP));
-        expectLastCall();
-        replay(mockedKeyValueTimestampStore);
-
         keyValueStoreFacade.putAll(asList(
             KeyValue.pair("key1", "value1"),
             KeyValue.pair("key2", "value2")
         ));
-        verify(mockedKeyValueTimestampStore);
+        verify(mockedKeyValueTimestampStore)
+            .put("key1", ValueAndTimestamp.make("value1", ConsumerRecord.NO_TIMESTAMP));
+        verify(mockedKeyValueTimestampStore)
+            .put("key2", ValueAndTimestamp.make("value2", ConsumerRecord.NO_TIMESTAMP));
     }
 
     @Test
     public void shouldDeleteAndReturnPlainValue() {
-        expect(mockedKeyValueTimestampStore.delete("key"))
-            .andReturn(null)
-            .andReturn(ValueAndTimestamp.make("oldValue", 42L));
-        replay(mockedKeyValueTimestampStore);
+        doReturn(null, ValueAndTimestamp.make("oldValue", 42L))
+            .when(mockedKeyValueTimestampStore).delete("key");
 
         assertNull(keyValueStoreFacade.delete("key"));
         assertThat(keyValueStoreFacade.delete("key"), is("oldValue"));
-        verify(mockedKeyValueTimestampStore);
+        verify(mockedKeyValueTimestampStore, times(2)).delete("key");
     }
 
     @Test
     public void shouldForwardFlush() {
-        mockedKeyValueTimestampStore.flush();
-        expectLastCall();
-        replay(mockedKeyValueTimestampStore);
-
         keyValueStoreFacade.flush();
-        verify(mockedKeyValueTimestampStore);
+        verify(mockedKeyValueTimestampStore).flush();
     }
 
     @Test
     public void shouldForwardClose() {
-        mockedKeyValueTimestampStore.close();
-        expectLastCall();
-        replay(mockedKeyValueTimestampStore);
-
         keyValueStoreFacade.close();
-        verify(mockedKeyValueTimestampStore);
+        verify(mockedKeyValueTimestampStore).close();
     }
 
     @Test
     public void shouldReturnName() {
-        expect(mockedKeyValueTimestampStore.name()).andReturn("name");
-        replay(mockedKeyValueTimestampStore);
+        when(mockedKeyValueTimestampStore.name()).thenReturn("name");
 
         assertThat(keyValueStoreFacade.name(), is("name"));
-        verify(mockedKeyValueTimestampStore);
+        verify(mockedKeyValueTimestampStore).name();
     }
 
     @Test
     public void shouldReturnIsPersistent() {
-        expect(mockedKeyValueTimestampStore.persistent())
-            .andReturn(true)
-            .andReturn(false);
-        replay(mockedKeyValueTimestampStore);
+        when(mockedKeyValueTimestampStore.persistent())
+            .thenReturn(true, false);
 
         assertThat(keyValueStoreFacade.persistent(), is(true));
         assertThat(keyValueStoreFacade.persistent(), is(false));
-        verify(mockedKeyValueTimestampStore);
+        verify(mockedKeyValueTimestampStore, times(2)).persistent();
     }
 
     @Test
     public void shouldReturnIsOpen() {
-        expect(mockedKeyValueTimestampStore.isOpen())
-            .andReturn(true)
-            .andReturn(false);
-        replay(mockedKeyValueTimestampStore);
+        when(mockedKeyValueTimestampStore.isOpen())
+            .thenReturn(true, false);
 
         assertThat(keyValueStoreFacade.isOpen(), is(true));
         assertThat(keyValueStoreFacade.isOpen(), is(false));
-        verify(mockedKeyValueTimestampStore);
+        verify(mockedKeyValueTimestampStore, times(2)).isOpen();
     }
 }
diff --git a/streams/test-utils/src/test/java/org/apache/kafka/streams/WindowStoreFacadeTest.java b/streams/test-utils/src/test/java/org/apache/kafka/streams/WindowStoreFacadeTest.java
index 434782c40b9..a0d575eedcd 100644
--- a/streams/test-utils/src/test/java/org/apache/kafka/streams/WindowStoreFacadeTest.java
+++ b/streams/test-utils/src/test/java/org/apache/kafka/streams/WindowStoreFacadeTest.java
@@ -17,26 +17,25 @@
 package org.apache.kafka.streams;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.streams.TopologyTestDriver.WindowStoreFacade;
 import org.apache.kafka.streams.processor.ProcessorContext;
 import org.apache.kafka.streams.processor.StateStore;
 import org.apache.kafka.streams.processor.StateStoreContext;
 import org.apache.kafka.streams.state.TimestampedWindowStore;
 import org.apache.kafka.streams.state.ValueAndTimestamp;
-import org.apache.kafka.streams.TopologyTestDriver.WindowStoreFacade;
-import org.easymock.EasyMock;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
-import static org.easymock.EasyMock.expect;
-import static org.easymock.EasyMock.expectLastCall;
-import static org.easymock.EasyMock.mock;
-import static org.easymock.EasyMock.replay;
-import static org.easymock.EasyMock.verify;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 public class WindowStoreFacadeTest {
-    private final TimestampedWindowStore<String, String> mockedWindowTimestampStore = EasyMock.mock(TimestampedWindowStore.class);
+    @SuppressWarnings("unchecked")
+    private final TimestampedWindowStore<String, String> mockedWindowTimestampStore = mock(TimestampedWindowStore.class);
 
     private WindowStoreFacade<String, String> windowStoreFacade;
 
@@ -50,87 +49,67 @@ public class WindowStoreFacadeTest {
     public void shouldForwardDeprecatedInit() {
         final ProcessorContext context = mock(ProcessorContext.class);
         final StateStore store = mock(StateStore.class);
-        mockedWindowTimestampStore.init(context, store);
-        expectLastCall();
-        replay(mockedWindowTimestampStore);
 
         windowStoreFacade.init(context, store);
-        verify(mockedWindowTimestampStore);
+        verify(mockedWindowTimestampStore)
+            .init(context, store);
     }
 
     @Test
     public void shouldForwardInit() {
         final StateStoreContext context = mock(StateStoreContext.class);
         final StateStore store = mock(StateStore.class);
-        mockedWindowTimestampStore.init(context, store);
-        expectLastCall();
-        replay(mockedWindowTimestampStore);
 
         windowStoreFacade.init(context, store);
-        verify(mockedWindowTimestampStore);
+        verify(mockedWindowTimestampStore)
+            .init(context, store);
     }
 
     @Test
     public void shouldPutWindowStartTimestampWithUnknownTimestamp() {
-        mockedWindowTimestampStore.put("key", ValueAndTimestamp.make("value", ConsumerRecord.NO_TIMESTAMP), 21L);
-        expectLastCall();
-        replay(mockedWindowTimestampStore);
-
         windowStoreFacade.put("key", "value", 21L);
-        verify(mockedWindowTimestampStore);
+        verify(mockedWindowTimestampStore)
+            .put("key", ValueAndTimestamp.make("value", ConsumerRecord.NO_TIMESTAMP), 21L);
     }
 
     @Test
     public void shouldForwardFlush() {
-        mockedWindowTimestampStore.flush();
-        expectLastCall();
-        replay(mockedWindowTimestampStore);
-
         windowStoreFacade.flush();
-        verify(mockedWindowTimestampStore);
+        verify(mockedWindowTimestampStore).flush();
     }
 
     @Test
     public void shouldForwardClose() {
-        mockedWindowTimestampStore.close();
-        expectLastCall();
-        replay(mockedWindowTimestampStore);
-
         windowStoreFacade.close();
-        verify(mockedWindowTimestampStore);
+        verify(mockedWindowTimestampStore).close();
     }
 
     @Test
     public void shouldReturnName() {
-        expect(mockedWindowTimestampStore.name()).andReturn("name");
-        replay(mockedWindowTimestampStore);
+        when(mockedWindowTimestampStore.name()).thenReturn("name");
 
         assertThat(windowStoreFacade.name(), is("name"));
-        verify(mockedWindowTimestampStore);
+        verify(mockedWindowTimestampStore).name();
     }
 
     @Test
     public void shouldReturnIsPersistent() {
-        expect(mockedWindowTimestampStore.persistent())
-            .andReturn(true)
-            .andReturn(false);
-        replay(mockedWindowTimestampStore);
+        when(mockedWindowTimestampStore.persistent())
+            .thenReturn(true, false);
 
         assertThat(windowStoreFacade.persistent(), is(true));
         assertThat(windowStoreFacade.persistent(), is(false));
-        verify(mockedWindowTimestampStore);
+        verify(mockedWindowTimestampStore, times(2)).persistent();
     }
 
     @Test
     public void shouldReturnIsOpen() {
-        expect(mockedWindowTimestampStore.isOpen())
-            .andReturn(true)
-            .andReturn(false);
-        replay(mockedWindowTimestampStore);
+        when(mockedWindowTimestampStore.isOpen())
+            .thenReturn(true, false);
 
         assertThat(windowStoreFacade.isOpen(), is(true));
         assertThat(windowStoreFacade.isOpen(), is(false));
-        verify(mockedWindowTimestampStore);
+        verify(mockedWindowTimestampStore, times(2)).isOpen();
     }
 
 }