Topic API control event ordering

[a1kaigorodov]

Problem

The Topic API client currently receives StartPartitionSessionEvent and StopPartitionSessionEvent events from AsyncReader without ordering guarantees. This leads to race conditions where a stop event may be processed before its corresponding start event, causing undefined behavior.

Expected Behavior

The Topic API should guarantee strict event ordering, ensuring that each control event is delivered to the client only after the previous event has been fully processed

[a1kaigorodov]

There is a related problem.
Currently, server reader sessions are closed before the onPartitionSessionClosed and onReaderClosed callbacks complete execution. This creates a race condition where partitions can be reassigned to other readers before the original reader has finished cleaning up its resources.

Example:

• Reader-1 is processing Partition 1
• Reader-1 is shutdown, triggering partition reassignment
• Reader-2 receives Partition 1 before Reader-1 completes cleanup

2025-11-06T19:57:27,052+03:00 [Test worker] INFO  y.c.t.t.p.TopicReaderPoolIntegrationTest:   : Before reader-1 shutdown
2025-11-06T19:57:27,052+03:00 [Test worker] INFO  t.y.t.r.i.ReaderImpl:   : [SSQSxd] Shutting down Reader
2025-11-06T19:57:27,052+03:00 [ForkJoinPool.commonPool-worker-1] INFO  t.y.t.r.i.ReadSession:   : Session shutdown
2025-11-06T19:57:27,052+03:00 [ForkJoinPool.commonPool-worker-1] INFO  t.y.t.r.i.ReadSession:   : Session stop
2025-11-06T19:57:27,052+03:00 [ForkJoinPool.commonPool-worker-1] DEBUG t.y.t.r.i.ReaderImpl:   : [SSQSxd.1] Session test-consumer_1_1_6911134001544361054_v1 onStop called
2025-11-06T19:57:27,052+03:00 [ForkJoinPool.commonPool-worker-1] INFO  t.y.t.r.i.PartitionSessionImpl:   : [SSQSxd.1/2-p1] Partition session for test-topic-507ba747-f83c-4ea5-8d93-2cadb4c6a4b9 is shutting down. Failing 0 commit futures...
2025-11-06T19:57:27,053+03:00 [reader-1-executor] INFO  y.c.t.t.p.TopicReaderPoolIntegrationTest:   : onPartitionSessionClosed PartitionSessionClosedEventImpl[partitionSession=tech.ydb.topic.read.PartitionSession@78ecd74d]
2025-11-06T19:57:27,053+03:00 [reader-1-executor] INFO  y.c.t.t.p.TopicReaderPoolIntegrationTest:   : before closing resources
2025-11-06T19:57:27,054+03:00 [grpc-default-executor-1] INFO  t.y.t.r.i.ReaderImpl:   : [SSQSxd.1] Session test-consumer_1_1_6911134001544361054_v1 closeDueToError called
2025-11-06T19:57:27,054+03:00 [grpc-default-executor-1] INFO  t.y.t.r.i.ReadSession:   : Session shutdown
2025-11-06T19:57:27,054+03:00 [grpc-default-executor-1] INFO  t.y.t.r.i.ReadSession:   : Session stop
2025-11-06T19:57:27,055+03:00 [grpc-default-executor-1] DEBUG t.y.t.r.i.ReadSession:   : Message received
2025-11-06T19:57:27,055+03:00 [grpc-default-executor-1] DEBUG t.y.t.r.i.ReaderImpl:   : [TRhnWI.1] processMessage called
2025-11-06T19:57:27,055+03:00 [grpc-default-executor-1] INFO  t.y.t.r.i.ReaderImpl:   : [TRhnWI.1/2-p1] Received StartPartitionSessionRequest for Topic "test-topic-507ba747-f83c-4ea5-8d93-2cadb4c6a4b9" and Consumer "test-consumer". Partition session 2 (partition 1) with committedOffset 0 and partitionOffsets [0-0)
2025-11-06T19:57:27,055+03:00 [grpc-default-executor-1] INFO  t.y.t.r.i.PartitionSessionImpl:   : [TRhnWI.1/2-p1] Partition session is started for Topic "test-topic-507ba747-f83c-4ea5-8d93-2cadb4c6a4b9" and Consumer "test-consumer". CommittedOffset: 0. Partition offsets: 0-0
2025-11-06T19:57:27,055+03:00 [reader-2-executor] INFO  y.c.t.t.p.TopicReaderPoolIntegrationTest:   : onStartPartitionSession StartPartitionSessionEventImpl[committedOffset=0,confirmCallback=tech.ydb.topic.read.impl.ReaderImpl$ReadSessionImpl$$Lambda/0x0000707b834e8ff8@6f3b6f8f,partitionOffsets=tech.ydb.topic.read.impl.OffsetsRangeImpl@3c1,partitionSession=tech.ydb.topic.read.PartitionSession@1b00d146]
2025-11-06T19:57:27,055+03:00 [reader-2-executor] INFO  t.y.t.r.i.ReaderImpl:   : [TRhnWI.1/2-p1] Sending StartPartitionSessionResponse with readOffset null and commitOffset null
2025-11-06T19:57:27,055+03:00 [reader-2-executor] DEBUG t.y.t.r.i.ReadSession:   : Sending request
2025-11-06T19:57:28,053+03:00 [reader-1-executor] INFO  y.c.t.t.p.TopicReaderPoolIntegrationTest:   : after closing resources
2025-11-06T19:57:28,053+03:00 [reader-1-executor] INFO  y.c.t.t.p.TopicReaderPoolIntegrationTest:   : onReaderClosed ReaderClosedEvent[]
2025-11-06T19:57:28,054+03:00 [Test worker] INFO  y.c.t.t.p.TopicReaderPoolIntegrationTest:   : After reader-1 shutdown

In the given example both reader-1 and reader-2 use single thread executors to handle events