From f1e5fc296166b651d3db28f285cdc98fe3b21bfa Mon Sep 17 00:00:00 2001 From: Irene Skvortsova Date: Thu, 25 Sep 2025 19:24:13 +0300 Subject: Add test on supportive partitions deletion for expired transactions (#25807) Co-authored-by: Irina Skvortsova --- ydb/core/persqueue/ut/pqtablet_ut.cpp | 35 +++++++++++++++++++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/ydb/core/persqueue/ut/pqtablet_ut.cpp b/ydb/core/persqueue/ut/pqtablet_ut.cpp index d3526699d4d..8ec2d433642 100644 --- a/ydb/core/persqueue/ut/pqtablet_ut.cpp +++ b/ydb/core/persqueue/ut/pqtablet_ut.cpp @@ -2495,6 +2495,41 @@ Y_UNIT_TEST_F(Kafka_Transaction_Supportive_Partitions_Should_Be_Deleted_After_Ti // wait till supportive partition for this kafka transaction is deleted WaitForExactSupportivePartitionsCount(0); +} + +Y_UNIT_TEST_F(Kafka_Transaction_Supportive_Partitions_Should_Be_Deleted_With_Delete_Partition_Done_Event_Drop, TPQTabletFixture) +{ + NKafka::TProducerInstanceId producerInstanceId = {1, 0}; + PQTabletPrepare({.partitions=1}, {}, *Ctx); + EnsurePipeExist(); + TString ownerCookie = CreateSupportivePartitionForKafka(producerInstanceId); + + // send data to create blobs for supportive partitions + SendKafkaTxnWriteRequest(producerInstanceId, ownerCookie); + + // validate supportive partition was created + WaitForExactSupportivePartitionsCount(1); + auto txInfo = GetTxWritesFromKV(); + UNIT_ASSERT_VALUES_EQUAL(txInfo.TxWritesSize(), 1); + UNIT_ASSERT_VALUES_EQUAL(txInfo.GetTxWrites(0).GetKafkaTransaction(), true); + + // increment time till after kafka txn timeout + ui64 kafkaTxnTimeoutMs = Ctx->Runtime->GetAppData(0).KafkaProxyConfig.GetTransactionTimeoutMs() + + KAFKA_TRANSACTION_DELETE_DELAY_MS; + Ctx->Runtime->AdvanceCurrentTime(TDuration::MilliSeconds(kafkaTxnTimeoutMs + 1)); + SendToPipe(Ctx->Edge, MakeHolder().Release()); + TAutoPtr deleteDoneEvent; + bool seenEvent = false; + // add observer for TEvPQ::TEvDeletePartitionDone request and skip it + AddOneTimeEventObserver(seenEvent, 1, [](TAutoPtr&) { + return TTestActorRuntimeBase::EEventAction::DROP; + }); + TDispatchOptions options; + options.CustomFinalCondition = [&seenEvent]() {return seenEvent;}; + UNIT_ASSERT(Ctx->Runtime->DispatchEvents(options)); + PQTabletRestart(*Ctx); + ResetPipe(); + // check that that our expired transaction has been deleted WaitForExactTxWritesCount(0); } -- cgit v1.3