Bugfix: error handling in internaltopicreader.addOnTransactionCompletedHandler is invalid (#1863)

Co-authored-by: Viktor Pentyukhov <nepunep@172.28.98.114-red.dhcp.yndx.net>

Author: 1NepuNep1

CHANGELOG.md

@@ -1,3 +1,5 @@
+* Fixed error handling in internaltopicreader.addOnTransactionCompletedHandler
+
 ## v3.116.3
 * Default grpc message size adjusted to the server's defaults
 

internal/topic/topicreaderinternal/stream_reader_impl.go

@@ -255,7 +255,7 @@ func (r *topicStreamReaderImpl) commitWithTransaction(
 		return err
 	})
 	if updateOffsetInTransactionErr == nil {
-		r.addOnTransactionCompletedHandler(ctx, tx, batch, updateOffsetInTransactionErr)
+		r.addOnTransactionCompletedHandler(ctx, tx, batch)
 	} else {
 		_ = retry.Retry(ctx, func(ctx context.Context) (err error) {
 			defer func() {
@@ -287,7 +287,6 @@ func (r *topicStreamReaderImpl) addOnTransactionCompletedHandler(
 	ctx context.Context,
 	tx tx.Transaction,
 	batch *topicreadercommon.PublicBatch,
-	updateOffesetInTransactionErr error,
 ) {
 	commitRange := topicreadercommon.GetCommitRange(batch)
 	tx.OnCompleted(func(transactionResult error) {
@@ -308,7 +307,7 @@ func (r *topicStreamReaderImpl) addOnTransactionCompletedHandler(
 			topicreadercommon.BatchGetPartitionSession(batch).SetCommittedOffsetForward(commitRange.CommitOffsetEnd)
 		} else {
 			_ = r.CloseWithError(xcontext.ValueOnly(ctx), xerrors.WithStackTrace(xerrors.Retryable(
-				fmt.Errorf("ydb: failed batch commit because transaction doesn't committed: %w", updateOffesetInTransactionErr),
+				fmt.Errorf("ydb: failed batch commit because transaction doesn't committed: %w", transactionResult),
 			)))
 		}
 	})