[-] too many transactions

Author: Alek5andr-Kotov

ydb/core/persqueue/pq_impl.cpp

@@ -41,6 +41,7 @@ static constexpr ui32 CACHE_SIZE = 100_MB;
 static constexpr ui32 MAX_BYTES = 25_MB;
 static constexpr ui32 MAX_SOURCE_ID_LENGTH = 2048;
 static constexpr ui32 MAX_HEARTBEAT_SIZE = 2_KB;
+static constexpr ui32 MAX_TXS = 1000;
 
 struct TChangeNotification {
  TChangeNotification(const TActorId& actor, const ui64 txId)
@@ -3274,10 +3275,6 @@ void TPersQueue::HandleDataTransaction(TAutoPtr<TEvPersQueue::TEvProposeTransact
  return;
  }
 
- //
- // TODO(abcdef): сохранить пока инициализируемся. TEvPersQueue::TEvHasDataInfo::TPtr как образец. не только конфиг. Inited==true
- //
-
  if (txBody.OperationsSize() <= 0) {
  PQ_LOG_D("TxId " << event.GetTxId() << " empty list of operations");
  SendProposeTransactionAbort(ActorIdFromProto(event.GetSourceActor()),
@@ -3346,7 +3343,6 @@ void TPersQueue::HandleDataTransaction(TAutoPtr<TEvPersQueue::TEvProposeTransact
  return;
  }
 
-
  if (txBody.GetImmediate()) {
  PQ_LOG_D("immediate transaction");
  TPartitionId originalPartitionId(txBody.GetOperations(0).GetPartitionId());
@@ -3360,6 +3356,15 @@ void TPersQueue::HandleDataTransaction(TAutoPtr<TEvPersQueue::TEvProposeTransact
 
  ctx.Send(partition.Actor, ev.Release());
  } else {
+ if ((EvProposeTransactionQueue.size() + Txs.size()) >= MAX_TXS) {
+ SendProposeTransactionAbort(ActorIdFromProto(event.GetSourceActor()),
+ event.GetTxId(),
+ NKikimrPQ::TError::ERROR,
+ "too many transactions",
+ ctx);
+ return;
+ }
+
  PQ_LOG_D("distributed transaction");
  EvProposeTransactionQueue.emplace_back(ev.Release());
 

ydb/core/persqueue/ut/pqtablet_ut.cpp

@@ -2107,6 +2107,47 @@ Y_UNIT_TEST_F(TEvReadSet_For_A_Non_Existent_Tablet, TPQTabletFixture)
  WaitForTheTransactionToBeDeleted(txId);
 }
 
+Y_UNIT_TEST_F(Limit_On_The_Number_Of_Transactons, TPQTabletFixture)
+{
+ const ui64 mockTabletId = MakeTabletID(false, 22222);
+ const ui64 txId = 67890;
+
+ PQTabletPrepare({.partitions=1}, {}, *Ctx);
+
+ for (ui64 i = 0; i < 1002; ++i) {
+ SendProposeTransactionRequest({.TxId=txId + i,
+ .Senders={mockTabletId}, .Receivers={mockTabletId},
+ .TxOps={
+ {.Partition=0, .Consumer="user", .Begin=0, .End=0, .Path="/topic"},
+ }});
+ }
+
+ size_t preparedCount = 0;
+ size_t abortedCount = 0;
+
+ for (ui64 i = 0; i < 1002; ++i) {
+ auto event = Ctx->Runtime->GrabEdgeEvent<TEvPersQueue::TEvProposeTransactionResult>();
+ UNIT_ASSERT(event != nullptr);
+
+ UNIT_ASSERT(event->Record.HasStatus());
+
+ const auto status = event->Record.GetStatus();
+ switch (status) {
+ case NKikimrPQ::TEvProposeTransactionResult::PREPARED:
+ ++preparedCount;
+ break;
+ case NKikimrPQ::TEvProposeTransactionResult::ABORTED:
+ ++abortedCount;
+ break;
+ default:
+ UNIT_FAIL("unexpected transaction status " << NKikimrPQ::TEvProposeTransactionResult_EStatus_Name(status));
+ }
+ }
+
+ UNIT_ASSERT_EQUAL(preparedCount, 1000);
+ UNIT_ASSERT_EQUAL(abortedCount, 2);
+}
+
 }
 
 }