diff --git a/tarantool-core/pom.xml b/tarantool-core/pom.xml
index 7096902a..71c2e09f 100644
--- a/tarantool-core/pom.xml
+++ b/tarantool-core/pom.xml
@@ -72,6 +72,11 @@
io.tarantool
testcontainers
+
+ org.mockito
+ mockito-junit-jupiter
+ test
+
diff --git a/tarantool-core/src/main/java/io/tarantool/core/state/AbstractDrainLoopMachine.java b/tarantool-core/src/main/java/io/tarantool/core/state/AbstractDrainLoopMachine.java
new file mode 100644
index 00000000..ed3a423c
--- /dev/null
+++ b/tarantool-core/src/main/java/io/tarantool/core/state/AbstractDrainLoopMachine.java
@@ -0,0 +1,57 @@
+/*
+ * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY
+ * All Rights Reserved.
+ */
+
+package io.tarantool.core.state;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class AbstractDrainLoopMachine implements StateMachine {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(AbstractDrainLoopMachine.class);
+
+ private final Queue events = new ConcurrentLinkedQueue<>();
+ private final AtomicInteger wip = new AtomicInteger(0);
+
+ @Override
+ public void handle(E event) {
+ this.events.offer(event);
+
+ if (this.wip.getAndIncrement() != 0) {
+ return;
+ }
+
+ int missed = 1;
+ try {
+ do {
+ E e;
+ while ((e = this.events.poll()) != null) {
+ try {
+ processEvent(e);
+ } catch (Throwable ex) {
+ LOGGER.error("[{}]: unexpected error in event handler. Try handle error...", this, ex);
+ try {
+ onHandleError(ex);
+ } catch (Throwable failSafeEx) {
+ LOGGER.error("[{}]: fail-safe handler failed", this, failSafeEx);
+ }
+ }
+ }
+ missed = this.wip.addAndGet(-missed);
+ } while (missed != 0);
+ } catch (Throwable criticalJvmError) {
+ this.wip.set(0);
+ throw criticalJvmError;
+ }
+ }
+
+ protected abstract void processEvent(E event);
+
+ protected abstract void onHandleError(Throwable error);
+}
diff --git a/tarantool-core/src/main/java/io/tarantool/core/state/AbstractNettyMachine.java b/tarantool-core/src/main/java/io/tarantool/core/state/AbstractNettyMachine.java
new file mode 100644
index 00000000..59942756
--- /dev/null
+++ b/tarantool-core/src/main/java/io/tarantool/core/state/AbstractNettyMachine.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY
+ * All Rights Reserved.
+ */
+
+package io.tarantool.core.state;
+
+import java.util.concurrent.RejectedExecutionException;
+
+import io.netty.channel.EventLoop;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class AbstractNettyMachine implements StateMachine {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(AbstractNettyMachine.class);
+
+ protected final EventLoop loop;
+
+ protected AbstractNettyMachine(EventLoop loop) {
+ this.loop = loop;
+ }
+
+ @Override
+ public void handle(E event) {
+ if (this.loop.inEventLoop()) {
+ internalHandle(event);
+ } else {
+ try {
+ this.loop.execute(() -> internalHandle(event));
+ } catch (RejectedExecutionException ex) {
+ LOGGER.error(
+ "[{}]: catch an unexpected machine event execution. Try handle this...", this, ex);
+ try {
+ onHandleRejectedException(event, ex);
+ } catch (Throwable failSafeEx) {
+ LOGGER.error("[{}]: fail-safe handler failed", this, failSafeEx);
+ }
+ }
+ }
+ }
+
+ private void internalHandle(E event) {
+ try {
+ processEvent(event);
+ } catch (Throwable error) {
+ try {
+ LOGGER.error("[{}]: unexpected error in event handler. Try handle error...", this, error);
+ onHandleError(error);
+ } catch (Throwable failSafeEx) {
+ LOGGER.error("[{}]: fail-safe handler failed", this, failSafeEx);
+ }
+ }
+ }
+
+ protected abstract void processEvent(E event);
+
+ protected abstract void onHandleError(Throwable error);
+
+ protected abstract void onHandleRejectedException(E event, RejectedExecutionException e);
+}
diff --git a/tarantool-core/src/main/java/io/tarantool/core/state/Event.java b/tarantool-core/src/main/java/io/tarantool/core/state/Event.java
new file mode 100644
index 00000000..eb53520f
--- /dev/null
+++ b/tarantool-core/src/main/java/io/tarantool/core/state/Event.java
@@ -0,0 +1,8 @@
+/*
+ * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY
+ * All Rights Reserved.
+ */
+
+package io.tarantool.core.state;
+
+public interface Event {}
diff --git a/tarantool-core/src/main/java/io/tarantool/core/state/State.java b/tarantool-core/src/main/java/io/tarantool/core/state/State.java
new file mode 100644
index 00000000..dea426eb
--- /dev/null
+++ b/tarantool-core/src/main/java/io/tarantool/core/state/State.java
@@ -0,0 +1,11 @@
+/*
+ * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY
+ * All Rights Reserved.
+ */
+
+package io.tarantool.core.state;
+
+public interface State {
+
+ void handle(E event);
+}
diff --git a/tarantool-core/src/main/java/io/tarantool/core/state/StateMachine.java b/tarantool-core/src/main/java/io/tarantool/core/state/StateMachine.java
new file mode 100644
index 00000000..181bf99f
--- /dev/null
+++ b/tarantool-core/src/main/java/io/tarantool/core/state/StateMachine.java
@@ -0,0 +1,11 @@
+/*
+ * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY
+ * All Rights Reserved.
+ */
+
+package io.tarantool.core.state;
+
+public interface StateMachine {
+
+ void handle(E event);
+}
diff --git a/tarantool-core/src/test/java/io/tarantool/core/state/AbstractNettyMachineTest.java b/tarantool-core/src/test/java/io/tarantool/core/state/AbstractNettyMachineTest.java
new file mode 100644
index 00000000..d3cb2f22
--- /dev/null
+++ b/tarantool-core/src/test/java/io/tarantool/core/state/AbstractNettyMachineTest.java
@@ -0,0 +1,151 @@
+/*
+ * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY
+ * All Rights Reserved.
+ */
+
+package io.tarantool.core.state;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.RejectedExecutionException;
+
+import io.netty.channel.EventLoop;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+class AbstractNettyMachineTest {
+
+ private EventLoop mockLoop;
+ private TestNettyMachine machine;
+
+ @BeforeEach
+ void setUp() {
+ mockLoop = Mockito.mock(EventLoop.class);
+ machine = new TestNettyMachine(mockLoop);
+ }
+
+ @Test
+ void testHandleInEventLoop() {
+ Mockito.when(mockLoop.inEventLoop()).thenReturn(true);
+ final Event event = Mockito.mock(Event.class);
+
+ machine.handle(event);
+
+ Assertions.assertEquals(1, machine.processedEvents.size());
+ Assertions.assertSame(event, machine.processedEvents.get(0));
+ Mockito.verify(mockLoop, Mockito.never()).execute(Mockito.any(Runnable.class));
+ }
+
+ @Test
+ void testHandleOutsideEventLoop() {
+ Mockito.when(mockLoop.inEventLoop()).thenReturn(false);
+ final Event event = Mockito.mock(Event.class);
+
+ machine.handle(event);
+
+ Assertions.assertEquals(0, machine.processedEvents.size());
+
+ final ArgumentCaptor runnableCaptor = ArgumentCaptor.forClass(Runnable.class);
+ Mockito.verify(mockLoop, Mockito.times(1)).execute(runnableCaptor.capture());
+
+ runnableCaptor.getValue().run();
+
+ Assertions.assertEquals(1, machine.processedEvents.size());
+ Assertions.assertSame(event, machine.processedEvents.get(0));
+ }
+
+ @Test
+ void testRecoveryAfterThrowable() {
+ Mockito.when(mockLoop.inEventLoop()).thenReturn(true);
+ final Event event = Mockito.mock(Event.class);
+ final RuntimeException expectedException = new RuntimeException("Сбой бизнес-логики");
+
+ machine.onProcessHook =
+ () -> {
+ throw expectedException;
+ };
+
+ Assertions.assertDoesNotThrow(() -> machine.handle(event));
+
+ Assertions.assertEquals(1, machine.caughtErrors.size());
+ Assertions.assertSame(expectedException, machine.caughtErrors.get(0));
+ Assertions.assertEquals(0, machine.processedEvents.size());
+ }
+
+ @Test
+ void testHandleRejectedExecutionException() {
+ Mockito.when(mockLoop.inEventLoop()).thenReturn(false);
+ Mockito.doThrow(new RejectedExecutionException("Loop is shutdown"))
+ .when(mockLoop)
+ .execute(Mockito.any(Runnable.class));
+
+ final Event event = Mockito.mock(Event.class);
+
+ Assertions.assertDoesNotThrow(() -> machine.handle(event));
+
+ // Проверяем, что сработал наш новый безопасный обработчик отказа и контекст события НЕ потерян
+ Assertions.assertEquals(1, machine.rejectedEvents.size());
+ Assertions.assertSame(event, machine.rejectedEvents.get(0));
+ Assertions.assertEquals(0, machine.processedEvents.size());
+ }
+
+ @Test
+ void testFailSafeHandlerProtectsAgainstDoubleFault() {
+ Mockito.when(mockLoop.inEventLoop()).thenReturn(true);
+ final Event event = Mockito.mock(Event.class);
+
+ machine.onProcessHook =
+ () -> {
+ throw new RuntimeException("First error");
+ };
+ TestNettyMachine brokenMachine =
+ new TestNettyMachine(mockLoop) {
+ @Override
+ protected void processEvent(Event event) {
+ throw new RuntimeException("First error");
+ }
+
+ @Override
+ protected void onHandleError(Throwable error) {
+ throw new RuntimeException("Second error onError inside");
+ }
+ };
+
+ Assertions.assertDoesNotThrow(() -> brokenMachine.handle(event));
+ }
+
+ private static class TestNettyMachine extends AbstractNettyMachine {
+
+ final List processedEvents = Collections.synchronizedList(new ArrayList<>());
+ final List caughtErrors = Collections.synchronizedList(new ArrayList<>());
+ final List rejectedEvents = Collections.synchronizedList(new ArrayList<>());
+
+ volatile Runnable onProcessHook = null;
+
+ protected TestNettyMachine(EventLoop loop) {
+ super(loop);
+ }
+
+ @Override
+ protected void processEvent(Event event) {
+ if (onProcessHook != null) {
+ onProcessHook.run();
+ }
+ processedEvents.add(event);
+ }
+
+ @Override
+ protected void onHandleError(Throwable error) {
+ caughtErrors.add(error);
+ }
+
+ @Override
+ protected void onHandleRejectedException(Event event, RejectedExecutionException e) {
+ rejectedEvents.add(event);
+ }
+ }
+}
diff --git a/tarantool-core/src/test/java/io/tarantool/core/unit/state/AbstractDrainLoopMachineTest.java b/tarantool-core/src/test/java/io/tarantool/core/unit/state/AbstractDrainLoopMachineTest.java
new file mode 100644
index 00000000..eb70a7e2
--- /dev/null
+++ b/tarantool-core/src/test/java/io/tarantool/core/unit/state/AbstractDrainLoopMachineTest.java
@@ -0,0 +1,176 @@
+/*
+ * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY
+ * All Rights Reserved.
+ */
+
+package io.tarantool.core.unit.state;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import io.tarantool.core.state.AbstractDrainLoopMachine;
+import io.tarantool.core.state.Event;
+
+class AbstractDrainLoopMachineTest {
+
+ private TestStateMachine stateMachine;
+
+ @BeforeEach
+ void setUp() {
+ stateMachine = new TestStateMachine();
+ }
+
+ @Test
+ void testSequentialProcessing() {
+ final Event event1 = Mockito.mock(Event.class);
+ final Event event2 = Mockito.mock(Event.class);
+
+ this.stateMachine.handle(event1);
+ this.stateMachine.handle(event2);
+
+ final List processed = this.stateMachine.processedEvents();
+ Assertions.assertEquals(2, processed.size());
+ Assertions.assertSame(event1, processed.get(0));
+ Assertions.assertSame(event2, processed.get(1));
+ Assertions.assertEquals(0, this.stateMachine.errorCount());
+ }
+
+ @Test
+ void testRecoveryAfterException() {
+ final Event event1 = Mockito.mock(Event.class);
+ final Event event2 = Mockito.mock(Event.class);
+ final Event event3 = Mockito.mock(Event.class);
+
+ this.stateMachine.handle(event1);
+ this.stateMachine.throwOnEvent = new RuntimeException("Simulation error");
+
+ this.stateMachine.handle(event2);
+ this.stateMachine.handle(event3);
+
+ Assertions.assertEquals(1, this.stateMachine.errorCount());
+ final List processed = this.stateMachine.processedEvents();
+ Assertions.assertEquals(2, processed.size());
+ Assertions.assertSame(event1, processed.get(0));
+ Assertions.assertSame(event3, processed.get(1));
+ }
+
+ @Test
+ void testConcurrentDrainLoop() throws InterruptedException {
+ int threadCount = 10;
+ int eventsPerThread = 10000;
+ int totalEvents = threadCount * eventsPerThread;
+
+ final ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+ final CountDownLatch startLatch = new CountDownLatch(1);
+ final CountDownLatch finishLatch = new CountDownLatch(threadCount);
+
+ // Генерируем пачку потоков, которые будут одновременно бомбардировать машину событиями
+ for (int i = 0; i < threadCount; i++) {
+ executor.submit(
+ () -> {
+ try {
+ startLatch.await();
+ for (int j = 0; j < eventsPerThread; j++) {
+ this.stateMachine.handle(Mockito.mock(Event.class));
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ finishLatch.countDown();
+ }
+ });
+ }
+
+ startLatch.countDown(); // Даем старт всем потокам одновременно
+ boolean completed = finishLatch.await(5, TimeUnit.SECONDS);
+ executor.shutdown();
+
+ Assertions.assertTrue(completed);
+ Assertions.assertEquals(totalEvents, stateMachine.processedEvents().size());
+ Assertions.assertEquals(0, stateMachine.errorCount());
+ }
+
+ @Test
+ void testEventSerialization() throws InterruptedException {
+ final Event event1 = Mockito.mock(Event.class);
+ final Event event2 = Mockito.mock(Event.class);
+
+ final CountDownLatch inProcessEvent1 = new CountDownLatch(1);
+ final CountDownLatch proceedEvent1 = new CountDownLatch(1);
+ final AtomicInteger activeThreadsInHandler = new AtomicInteger(0);
+
+ this.stateMachine.onProcessHook =
+ () -> {
+ activeThreadsInHandler.incrementAndGet();
+ inProcessEvent1.countDown();
+ try {
+ proceedEvent1.await(2, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ activeThreadsInHandler.decrementAndGet();
+ }
+ };
+
+ final Thread t1 = new Thread(() -> stateMachine.handle(event1));
+ t1.start();
+
+ Assertions.assertTrue(inProcessEvent1.await(1, TimeUnit.SECONDS));
+
+ final long startTime = System.currentTimeMillis();
+ this.stateMachine.handle(event2);
+ final long duration = System.currentTimeMillis() - startTime;
+
+ Assertions.assertTrue(duration < 500);
+
+ proceedEvent1.countDown();
+ t1.join(2000);
+
+ Assertions.assertEquals(2, stateMachine.processedEvents().size());
+ }
+
+ private static class TestStateMachine extends AbstractDrainLoopMachine {
+
+ private final List processedEvents = Collections.synchronizedList(new ArrayList<>());
+ private final AtomicInteger errorCount = new AtomicInteger(0);
+ private volatile Runnable onProcessHook = null;
+ private volatile RuntimeException throwOnEvent = null;
+
+ @Override
+ protected void processEvent(Event event) {
+ if (throwOnEvent != null) {
+ final RuntimeException ex = throwOnEvent;
+ throwOnEvent = null;
+ throw ex;
+ }
+ if (onProcessHook != null) {
+ onProcessHook.run();
+ }
+ processedEvents.add(event);
+ }
+
+ @Override
+ protected void onHandleError(Throwable error) {
+ this.errorCount.incrementAndGet();
+ }
+
+ public List processedEvents() {
+ return this.processedEvents;
+ }
+
+ public int errorCount() {
+ return this.errorCount.get();
+ }
+ }
+}
diff --git a/tarantool-java-sdk-bom/pom.xml b/tarantool-java-sdk-bom/pom.xml
index c0c2be06..171f1827 100644
--- a/tarantool-java-sdk-bom/pom.xml
+++ b/tarantool-java-sdk-bom/pom.xml
@@ -34,6 +34,7 @@
5.6.0
1.3.3
1.67.0
+ 4.8.1
@@ -242,6 +243,14 @@
${instancio.version}
test
+
+
+ org.mockito
+ mockito-bom
+ ${mockito.version}
+ pom
+ import
+