From 14faff28dbb0255fe9ff452e15dee6b37c2d2e64 Mon Sep 17 00:00:00 2001 From: Ilnaz Nizametdinov Date: Wed, 2 Oct 2024 21:42:12 +0300 Subject: [PATCH] Better handling of connection loss (#9993) --- .../change_sender_common_ops.h | 20 +++++++++++---- ydb/core/change_exchange/util.cpp | 15 +++++++++++ ydb/core/change_exchange/util.h | 9 +++++++ ydb/core/change_exchange/ya.make | 1 + .../datashard/change_sender_async_index.cpp | 20 ++++++--------- .../tx/datashard/change_sender_cdc_stream.cpp | 25 ++++++++----------- .../replication/service/table_writer_impl.h | 21 ++++++---------- 7 files changed, 65 insertions(+), 46 deletions(-) create mode 100644 ydb/core/change_exchange/util.cpp create mode 100644 ydb/core/change_exchange/util.h diff --git a/ydb/core/change_exchange/change_sender_common_ops.h b/ydb/core/change_exchange/change_sender_common_ops.h index 823b208f5323..7c472e704443 100644 --- a/ydb/core/change_exchange/change_sender_common_ops.h +++ b/ydb/core/change_exchange/change_sender_common_ops.h @@ -413,20 +413,30 @@ class TBaseChangeSender { } TActorId GetChangeServer() const { return ChangeServer; } - void CreateSenders(const TVector& partitionIds, bool partitioningChanged = true) { - if (partitioningChanged) { + +private: + void CreateSendersImpl(const TVector& partitionIds) { + if (partitionIds) { CreateMissingSenders(partitionIds); } else { - RecreateSenders(GonePartitions); + RecreateSenders(std::exchange(GonePartitions, {})); } - GonePartitions.clear(); - if (!Enqueued || !RequestRecords()) { SendRecords(); } } +protected: + void CreateSenders(const TVector& partitionIds) { + Y_ABORT_UNLESS(partitionIds); + CreateSendersImpl(partitionIds); + } + + void CreateSenders() { + CreateSendersImpl({}); + } + void KillSenders() { for (const auto& [_, sender] : std::exchange(Senders, {})) { if (sender.ActorId) { diff --git a/ydb/core/change_exchange/util.cpp b/ydb/core/change_exchange/util.cpp new file mode 100644 index 000000000000..c4c0516e0e94 --- /dev/null +++ b/ydb/core/change_exchange/util.cpp @@ -0,0 +1,15 @@ +#include "util.h" + +namespace NKikimr::NChangeExchange { + +TVector MakePartitionIds(const TVector& partitions) { + TVector result(::Reserve(partitions.size())); + + for (const auto& partition : partitions) { + result.push_back(partition.ShardId); + } + + return result; +} + +} diff --git a/ydb/core/change_exchange/util.h b/ydb/core/change_exchange/util.h new file mode 100644 index 000000000000..f8ba146fdeaf --- /dev/null +++ b/ydb/core/change_exchange/util.h @@ -0,0 +1,9 @@ +#pragma once + +#include + +namespace NKikimr::NChangeExchange { + +TVector MakePartitionIds(const TVector& partitions); + +} diff --git a/ydb/core/change_exchange/ya.make b/ydb/core/change_exchange/ya.make index b95ab2178442..680c246118ea 100644 --- a/ydb/core/change_exchange/ya.make +++ b/ydb/core/change_exchange/ya.make @@ -4,6 +4,7 @@ SRCS( change_exchange.cpp change_record.cpp change_sender_monitoring.cpp + util.cpp ) GENERATE_ENUM_SERIALIZATION(change_record.h) diff --git a/ydb/core/tx/datashard/change_sender_async_index.cpp b/ydb/core/tx/datashard/change_sender_async_index.cpp index 38492b20728e..106d03406b58 100644 --- a/ydb/core/tx/datashard/change_sender_async_index.cpp +++ b/ydb/core/tx/datashard/change_sender_async_index.cpp @@ -6,6 +6,7 @@ #include #include #include +#include #include #include #include @@ -435,16 +436,6 @@ class TAsyncIndexChangeSenderMain return Check(&TSchemeCacheHelpers::CheckEntryKind, &TThis::LogWarnAndRetry, entry, expected); } - static TVector MakePartitionIds(const TVector& partitions) { - TVector result(Reserve(partitions.size())); - - for (const auto& partition : partitions) { - result.push_back(partition.ShardId); // partition = shard - } - - return result; - } - /// ResolveUserTable void ResolveUserTable() { @@ -611,6 +602,11 @@ class TAsyncIndexChangeSenderMain return; } + if (IndexTableVersion && IndexTableVersion == entry.Self->Info.GetVersion().GetGeneralVersion()) { + CreateSenders(); + return Become(&TThis::StateMain); + } + TagMap.clear(); TVector keyColumnTypes; @@ -692,11 +688,9 @@ class TAsyncIndexChangeSenderMain return Retry(); } - const bool versionChanged = !IndexTableVersion || IndexTableVersion != entry.GeneralVersion; IndexTableVersion = entry.GeneralVersion; - KeyDesc = std::move(entry.KeyDescription); - CreateSenders(MakePartitionIds(KeyDesc->GetPartitions()), versionChanged); + CreateSenders(NChangeExchange::MakePartitionIds(KeyDesc->GetPartitions())); Become(&TThis::StateMain); } diff --git a/ydb/core/tx/datashard/change_sender_cdc_stream.cpp b/ydb/core/tx/datashard/change_sender_cdc_stream.cpp index a605c1cc7099..980c269f15ca 100644 --- a/ydb/core/tx/datashard/change_sender_cdc_stream.cpp +++ b/ydb/core/tx/datashard/change_sender_cdc_stream.cpp @@ -6,6 +6,7 @@ #include #include +#include #include #include #include @@ -390,16 +391,6 @@ class TCdcChangeSenderMain return false; } - static TVector MakePartitionIds(const TVector& partitions) { - TVector result(Reserve(partitions.size())); - - for (const auto& partition : partitions) { - result.push_back(partition.ShardId); - } - - return result; - } - /// ResolveCdcStream void ResolveCdcStream() { @@ -521,6 +512,14 @@ class TCdcChangeSenderMain return; } + const auto topicVersion = entry.Self->Info.GetVersion().GetGeneralVersion(); + if (TopicVersion && TopicVersion == topicVersion) { + CreateSenders(); + return Become(&TThis::StateMain); + } + + TopicVersion = topicVersion; + const auto& pqDesc = entry.PQGroupInfo->Description; const auto& pqConfig = pqDesc.GetPQTabletConfig(); @@ -529,16 +528,12 @@ class TCdcChangeSenderMain PartitionToShard.emplace(partition.GetPartitionId(), partition.GetTabletId()); } - const auto topicVersion = entry.Self->Info.GetVersion().GetGeneralVersion(); - const bool versionChanged = !TopicVersion || TopicVersion != topicVersion; - TopicVersion = topicVersion; - Y_ABORT_UNLESS(entry.PQGroupInfo->Schema); KeyDesc = NKikimr::TKeyDesc::CreateMiniKeyDesc(entry.PQGroupInfo->Schema); Y_ABORT_UNLESS(entry.PQGroupInfo->Partitioning); KeyDesc->Partitioning = std::make_shared>(entry.PQGroupInfo->Partitioning); - CreateSenders(MakePartitionIds(*KeyDesc->Partitioning), versionChanged); + CreateSenders(NChangeExchange::MakePartitionIds(*KeyDesc->Partitioning)); Become(&TThis::StateMain); } diff --git a/ydb/core/tx/replication/service/table_writer_impl.h b/ydb/core/tx/replication/service/table_writer_impl.h index 1fd77232fb2d..2af13fa22042 100644 --- a/ydb/core/tx/replication/service/table_writer_impl.h +++ b/ydb/core/tx/replication/service/table_writer_impl.h @@ -6,6 +6,7 @@ #include #include +#include #include #include #include @@ -278,16 +279,6 @@ class TLocalTableWriter return Check(&TSchemeCacheHelpers::CheckEntryKind, &TThis::LogCritAndLeave, entry, expected); } - static TVector MakePartitionIds(const TVector& partitions) { - TVector result(::Reserve(partitions.size())); - - for (const auto& partition : partitions) { - result.push_back(partition.ShardId); - } - - return result; - } - void Registered(TActorSystem*, const TActorId&) override { this->ChangeServer = this->SelfId(); } @@ -348,6 +339,12 @@ class TLocalTableWriter return; } + if (TableVersion && TableVersion == entry.Self->Info.GetVersion().GetGeneralVersion()) { + Y_ABORT_UNLESS(Initialized); + Resolving = false; + return CreateSenders(); + } + auto schema = MakeIntrusive(); if (entry.Self && entry.Self->Info.HasVersion()) { schema->Version = entry.Self->Info.GetVersion().GetTableSchemaVersion(); @@ -415,11 +412,9 @@ class TLocalTableWriter return LogWarnAndRetry("Empty partitions"); } - const bool versionChanged = !TableVersion || TableVersion != entry.GeneralVersion; TableVersion = entry.GeneralVersion; - KeyDesc = std::move(entry.KeyDescription); - this->CreateSenders(MakePartitionIds(KeyDesc->GetPartitions()), versionChanged); + this->CreateSenders(NChangeExchange::MakePartitionIds(KeyDesc->GetPartitions())); if (!Initialized) { this->Send(Worker, new TEvWorker::TEvHandshake());