[+] Changes for Ya.Metrika #26244

Author: Alek5andr-Kotov

ydb/core/persqueue/partition.cpp

@@ -1466,7 +1466,7 @@ TPartition::EProcessResult TPartition::ApplyWriteInfoResponse(TTransaction& tx)
         if (auto inFlightIter = TxInflightMaxSeqNoPerSourceId.find(s.first); !inFlightIter.IsEnd()) {
             if (SeqnoViolation(inFlightIter->second.KafkaProducerEpoch, inFlightIter->second.SeqNo, s.second.ProducerEpoch, s.second.MinSeqNo)) {
                 tx.Predicate = false;
-                tx.Message = (MakeTxWriteErrorMessage(tx.GetTxId(), TopicName(), Partition, s.first, existing->second.SeqNo) <<
+                tx.Message = (MakeTxWriteErrorMessage(tx.GetTxId(), TopicName(), Partition, s.first, inFlightIter->second.SeqNo) <<
                               "MinSeqNo violation failure. " <<
                               "SeqNo " << s.second.MinSeqNo);
                 tx.WriteInfoApplied = true;
@@ -1477,7 +1477,9 @@ TPartition::EProcessResult TPartition::ApplyWriteInfoResponse(TTransaction& tx)
         if (auto existing = knownSourceIds.find(s.first); !existing.IsEnd()) {
             if (SeqnoViolation(existing->second.ProducerEpoch, existing->second.SeqNo, s.second.ProducerEpoch, s.second.MinSeqNo)) {
                 tx.Predicate = false;
-                tx.Message = TStringBuilder() << "MinSeqNo violation failure on " << s.first;
+                tx.Message = (MakeTxWriteErrorMessage(tx.GetTxId(), TopicName(), Partition, s.first, existing->second.SeqNo) <<
+                              "MinSeqNo violation failure. " <<
+                              "SeqNo " << s.second.MinSeqNo);
                 tx.WriteInfoApplied = true;
                 break;
             }

ydb/core/persqueue/partition_init.cpp

@@ -532,29 +532,26 @@ enum EKeyPosition {
 // Calculates the location of keys relative to each other
 static EKeyPosition KeyPosition(const TKey& lhs, const TKey& rhs)
 {
-    // Called from FilterBlobsMetaData. The keys are pre-sorted
-    Y_ABORT_UNLESS(lhs.GetOffset() <= rhs.GetOffset(),
-                   "lhs: %s, rhs: %s",
-                   lhs.ToString().data(), rhs.ToString().data());
-
     if (lhs.GetOffset() == rhs.GetOffset()) {
         if (lhs.GetPartNo() == rhs.GetPartNo()) {
-            Y_ABORT_UNLESS(lhs.GetCount() <= rhs.GetCount(),
-                           "lhs: %s, rhs: %s",
-                           lhs.ToString().data(), rhs.ToString().data());
-            return RhsContainsLhs;
+            if (lhs.GetCount() < rhs.GetCount()) {
+                return RhsContainsLhs;
+            } else if (lhs.GetCount() == rhs.GetCount()) {
+                if (lhs.GetInternalPartsCount() < rhs.GetInternalPartsCount()) {
+                    return RhsContainsLhs;
+                } else {
+                    return LhsContainsRhs;
+                }
+            } else {
+                return LhsContainsRhs;
+            }
+        } else if (lhs.GetPartNo() > rhs.GetPartNo()) {
+            return LhsContainsRhs;
+        } else {
+            return RhsAfterLhs;
         }
-
-        // case lhs.GetOffset() == rhs.GetOffset() && lhs.GetPartNo() < rhs.GetPartNo()
-        Y_ABORT_UNLESS(lhs.GetPartNo() + lhs.GetInternalPartsCount() <= rhs.GetPartNo(),
-                       "lhs: %s, rhs: %s",
-                       lhs.ToString().data(), rhs.ToString().data());
-
-        return RhsAfterLhs;
     }
 
-    // case lhs.GetOffset() < rhs.GetOffset()
-
     if (ui64 nextOffset = lhs.GetOffset() + lhs.GetCount(); nextOffset > rhs.GetOffset()) {
         return LhsContainsRhs;
     } else if (nextOffset == rhs.GetOffset()) {