Merge 55d1434 into 42c8024

Author: Alek5andr-Kotov

ydb/core/persqueue/pq_impl.cpp

@@ -4667,6 +4667,8 @@ void TPersQueue::TryStartTransaction(const TActorContext& ctx)
     Y_ABORT_UNLESS(next);
 
     CheckTxState(ctx, *next);
+
+    TryWriteTxs(ctx);
 }
 
 void TPersQueue::OnInitComplete(const TActorContext& ctx)

ydb/core/protos/feature_flags.proto

@@ -98,16 +98,16 @@ message TFeatureFlags {
     reserved 83; // EnableKqpDataQuerySourceRead
     optional bool EnableSmallDiskOptimization = 84 [default = true];
     optional bool EnableDataShardVolatileTransactions = 85 [default = false];
-    optional bool EnableTopicServiceTx = 86 [default = false];
+    optional bool EnableTopicServiceTx = 86 [default = true];
     optional bool EnableLLVMCache = 87 [default = false];
     optional bool EnableExternalDataSources = 88 [default = false];
     optional bool EnableTopicDiskSubDomainQuota = 89 [default = true];
     optional bool EnableSeparationComputeActorsFromRead = 90 [default = false];
-    optional bool EnablePQConfigTransactionsAtSchemeShard = 91 [default = false];
+    optional bool EnablePQConfigTransactionsAtSchemeShard = 91 [default = true];
     optional bool EnableScriptExecutionOperations = 92 [default = true];
     optional bool EnableImplicitQueryParameterTypes = 93 [default = false];
     optional bool EnableForceImmediateEffectsExecution = 94 [default = false];
-    optional bool EnableTopicSplitMerge = 95 [default = false];
+    optional bool EnableTopicSplitMerge = 95 [default = true];
     optional bool EnableChangefeedDynamoDBStreamsFormat = 96 [default = true];
     optional bool ForceColumnTablesCompositeMarks = 97 [default = false];
     optional bool EnableSubscriptionsInDiscovery = 98 [default = false];

ydb/core/tx/datashard/datashard_ut_change_exchange.cpp

@@ -1247,6 +1247,8 @@ Y_UNIT_TEST_SUITE(Cdc) {
 
             // get records
             {
+                WaitForDataRecords(client, shardIt);
+
                 auto res = client.GetRecords(shardIt).ExtractValueSync();
                 UNIT_ASSERT_C(res.IsSuccess(), res.GetIssues().ToString());
                 UNIT_ASSERT_VALUES_EQUAL(res.GetResult().records().size(), records.size());
@@ -1268,6 +1270,19 @@ Y_UNIT_TEST_SUITE(Cdc) {
             }
         }
 
+        static void WaitForDataRecords(TDataStreamsClient& client, const TString& shardIt) {
+            int n = 0;
+            for (; n < 100; ++n) {
+                auto res = client.GetRecords(shardIt).ExtractValueSync();
+                UNIT_ASSERT_C(res.IsSuccess(), res.GetIssues().ToString());
+                if (res.GetResult().records().size()) {
+                    break;
+                }
+                Sleep(TDuration::MilliSeconds(100));
+            }
+            UNIT_ASSERT_VALUES_UNEQUAL(n, 100);
+        }
+
         static void Write(const TShardedTableOptions& tableDesc, const TCdcStream& streamDesc) {
             TTestYdsEnv env(tableDesc, streamDesc);