Merge stable 24-4 to stream-nb-24-4

.github/workflows/docs_release.yaml

@@ -10,28 +10,62 @@ on:
   workflow_dispatch:
 
 jobs:
-  release:
+  build:
+    runs-on: ubuntu-latest
     concurrency:
       group: release-documentation-${{ github.ref }}
       cancel-in-progress: true
-    runs-on: ubuntu-latest
+    permissions: write-all
     steps:
       - name: Checkout
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
+      - name: Build
+        uses: diplodoc-platform/docs-build-action@v3
+        with:
+          revision: "${{ github.sha }}"
+          src-root: ${{ vars.SRC_ROOT }}
+  upload:
+    needs: build
+    runs-on: ubuntu-latest
+    concurrency:
+      group: release-documentation-${{ github.ref }}
+      cancel-in-progress: true
+    permissions: write-all
+    steps:
+      - name: Upload
+        uses: diplodoc-platform/docs-upload-action@v1
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          storage-bucket: "common/${{ secrets.DOCS_PROJECT_NAME }}"
+          storage-endpoint: ${{ vars.DOCS_AWS_ENDPOINT }}
+          storage-access-key-id: ${{ secrets.DOCS_AWS_KEY_ID }}
+          storage-secret-access-key: ${{ secrets.DOCS_AWS_SECRET_ACCESS_KEY }}
+          storage-region: ${{ vars.DOCS_AWS_REGION }}
+  release:
+    needs: upload
+    runs-on: ubuntu-latest
+    concurrency:
+      group: release-documentation-${{ github.ref }}
+      cancel-in-progress: true
+    steps:
       - name: Extract version
         shell: bash
         run: echo "version=${GITHUB_HEAD_REF:-${GITHUB_REF#refs/heads/}}" | sed -e 's|stable-|v|g' -e 's|-|.|g' >> $GITHUB_OUTPUT
         id: extract_version
+      - name: Set update-only-version
+        id: set-update-only-version
+        run: |
+          if [[ "${GITHUB_REF##*/}" == "main" ]]; then
+            echo "update_only_version=false" >> $GITHUB_ENV
+          else
+            echo "update_only_version=true" >> $GITHUB_ENV
+          fi
       - name: Release
-        uses: diplodoc-platform/docs-release-action@v1
+        uses: diplodoc-platform/docs-release-action@v2
         with:
           revision: "${{ github.sha }}"
           version: "${{ steps.extract_version.outputs.version }}"
-          project-name: ${{ secrets.DOCS_PROJECT_NAME }}
-          src-root: ${{ vars.SRC_ROOT }}
-          storage-bucket: ${{ secrets.DOCS_AWS_BUCKET }}
-          storage-endpoint: ${{ vars.DOCS_AWS_ENDPOINT }}
+          storage-bucket: ${{ secrets.DOCS_PROJECT_NAME }}
           storage-access-key-id: ${{ secrets.DOCS_AWS_KEY_ID }}
           storage-secret-access-key: ${{ secrets.DOCS_AWS_SECRET_ACCESS_KEY }}
-          storage-region: ${{ vars.DOCS_AWS_REGION }}
-          shared-storage-bucket: true
+          update-only-version: "${{ env.update_only_version }}"

ydb/apps/version/version_definition.cpp

@@ -3,20 +3,13 @@
 NKikimrConfig::TCurrentCompatibilityInfo NKikimr::TCompatibilityInfo::MakeCurrent() {
     using TCurrentConstructor = NKikimr::TCompatibilityInfo::TProtoConstructor::TCurrentCompatibilityInfo;
     using TVersionConstructor = NKikimr::TCompatibilityInfo::TProtoConstructor::TVersion;
-    using TCompatibilityRuleConstructor = NKikimr::TCompatibilityInfo::TProtoConstructor::TCompatibilityRule;
+    // using TCompatibilityRuleConstructor = NKikimr::TCompatibilityInfo::TProtoConstructor::TCompatibilityRule;
 
     return TCurrentConstructor{
         .Application = "ydb",
         .Version = TVersionConstructor{
             .Year = 24,
-            .Major = 3,
+            .Major = 4,
         },
-        .CanConnectTo = {
-            TCompatibilityRuleConstructor{
-                .Application = "nbs",
-                .LowerLimit = TVersionConstructor{ .Year = 23, .Major = 3 },
-                .UpperLimit = TVersionConstructor{ .Year = 24, .Major = 3 },
-            }
-        }
     }.ToPB();
-}
+}

ydb/core/base/hive.h

@@ -705,15 +705,25 @@ namespace NKikimr {
                 Record.SetStatus(status);
                 Record.SetStatusMessage(statusMessage);
             }
+
+            TString ToString() const override {
+                TStringStream str;
+                str << ToStringHeader() << "{Status: " << NKikimrProto::EReplyStatus_Name(Record.GetStatus()).data();
+                str << " TabletID: " << Record.GetTabletID();
+                str << " Message: " << Record.GetStatusMessage();
+                str << "}";
+                return str.Str();
+            }
         };
 
         struct TEvLockTabletExecutionLost : public TEventPB<TEvLockTabletExecutionLost,
                 NKikimrHive::TEvLockTabletExecutionLost, EvLockTabletExecutionLost>
         {
             TEvLockTabletExecutionLost() = default;
 
-            explicit TEvLockTabletExecutionLost(ui64 tabletId) {
+            explicit TEvLockTabletExecutionLost(ui64 tabletId, NKikimrHive::ELockLostReason reason) {
                 Record.SetTabletID(tabletId);
+                Record.SetReason(reason);
             }
         };
 

ydb/core/blobstorage/nodewarden/node_warden_group_resolver.cpp

@@ -336,9 +336,12 @@ namespace NKikimr::NStorage {
         if (const auto it = Groups.find(r.GetGroupId()); it != Groups.end() && it->second.Group) {
             record.MutableGroup()->CopyFrom(*it->second.Group);
         }
+        THashSet<TGroupId> groupsAdded;
         for (const auto& [key, value] : LocalVDisks) {
             if (const auto& r = value.RuntimeData; r && !r->DonorMode) {
-                record.AddStartedGroupIds(r->GroupInfo->GroupID.GetRawId());
+                if (const auto& groupId = r->GroupInfo->GroupID; groupsAdded.insert(groupId).second) {
+                    record.AddStartedGroupIds(groupId.GetRawId());
+                }
             }
         }
         Send(ev->Sender, res.release(), 0, ev->Cookie);

ydb/core/client/server/msgbus_server_pq_metacache.cpp

@@ -516,7 +516,11 @@ class TPersQueueMetaCacheActor : public TActorBootstrapped<TPersQueueMetaCacheAc
         auto inserted = DescribeTopicsWaiters.insert(std::make_pair(reqId, waiter)).second;
         Y_ABORT_UNLESS(inserted);
 
+        TMaybe<TString> db = {};
+
         for (const auto& [path, database] : waiter->GetTopics()) {
+            if (!db) db = database;
+            if (*db != database) db = "";
             auto split = NKikimr::SplitPath(path);
             TSchemeCacheNavigate::TEntry entry;
             if (!split.empty()) {
@@ -529,8 +533,9 @@ class TPersQueueMetaCacheActor : public TActorBootstrapped<TPersQueueMetaCacheAc
 
             schemeCacheRequest->ResultSet.emplace_back(std::move(entry));
         }
-
-        LOG_DEBUG_S(ctx, NKikimrServices::PQ_METACACHE, "send request for " << (waiter->Type == EWaiterType::DescribeAllTopics ? " all " : "") << waiter->GetTopics().size() << " topics, got " << DescribeTopicsWaiters.size() << " requests infly");
+        if (db) schemeCacheRequest->DatabaseName = *db;
+        LOG_DEBUG_S(ctx, NKikimrServices::PQ_METACACHE, "send request for " << (waiter->Type == EWaiterType::DescribeAllTopics ? " all " : "")
+                             << waiter->GetTopics().size() << " topics, got " << DescribeTopicsWaiters.size() << " requests infly, db = \"" << db << "\"");
 
         ctx.Send(SchemeCacheId, new TEvTxProxySchemeCache::TEvNavigateKeySet(schemeCacheRequest.release()));
     }

ydb/core/config/init/init_impl.h

@@ -315,6 +315,7 @@ struct TCommonAppOptions {
     TString GRpcPublicHost = "";
     ui32 GRpcPublicPort = 0;
     ui32 GRpcsPublicPort = 0;
+    ui32 KafkaPort = 0;
     TVector<TString> GRpcPublicAddressesV4;
     TVector<TString> GRpcPublicAddressesV6;
     TString GRpcPublicTargetNameOverride = "";
@@ -383,6 +384,7 @@ struct TCommonAppOptions {
         opts.AddLongOption("grpc-public-host", "set public gRPC host for discovery").RequiredArgument("HOST").StoreResult(&GRpcPublicHost);
         opts.AddLongOption("grpc-public-port", "set public gRPC port for discovery").RequiredArgument("PORT").StoreResult(&GRpcPublicPort);
         opts.AddLongOption("grpcs-public-port", "set public gRPC SSL port for discovery").RequiredArgument("PORT").StoreResult(&GRpcsPublicPort);
+        opts.AddLongOption("kafka-port", "enable kafka proxy to listen on port").OptionalArgument("PORT").StoreResult(&KafkaPort);
         opts.AddLongOption("grpc-public-address-v4", "set public ipv4 address for discovery").RequiredArgument("ADDR").EmplaceTo(&GRpcPublicAddressesV4);
         opts.AddLongOption("grpc-public-address-v6", "set public ipv6 address for discovery").RequiredArgument("ADDR").EmplaceTo(&GRpcPublicAddressesV6);
         opts.AddLongOption("grpc-public-target-name-override", "set public hostname override for TLS in discovery").RequiredArgument("HOST").StoreResult(&GRpcPublicTargetNameOverride);
@@ -597,6 +599,12 @@ struct TCommonAppOptions {
             }
             ConfigUpdateTracer.AddUpdate(NKikimrConsole::TConfigItem::GRpcConfigItem, TConfigItemInfo::EUpdateKind::UpdateExplicitly);
         }
+	if (KafkaPort) {
+            auto& conf = *appConfig.MutableKafkaProxyConfig();
+            conf.SetEnableKafkaProxy(true);
+            conf.SetListeningPort(KafkaPort);
+            ConfigUpdateTracer.AddUpdate(NKikimrConsole::TConfigItem::KafkaProxyConfigItem, TConfigItemInfo::EUpdateKind::UpdateExplicitly);
+        }
         for (const auto& addr : GRpcPublicAddressesV4) {
             appConfig.MutableGRpcConfig()->AddPublicAddressesV4(addr);
         }

ydb/core/discovery/discovery.cpp

@@ -46,6 +46,16 @@ namespace NDiscovery {
         return false;
     }
 
+    bool CheckEndpointId(const TString& endpointId, const NKikimrStateStorage::TEndpointBoardEntry &entry) {
+        if (endpointId.empty() && !entry.HasEndpointId())
+            return true;
+
+        if (entry.HasEndpointId() && entry.GetEndpointId() == endpointId)
+            return true;
+
+        return false;
+    }
+
     bool IsSafeLocationMarker(TStringBuf location) {
         const ui8* isrc = reinterpret_cast<const ui8*>(location.data());
         for (auto idx : xrange(location.size())) {
@@ -128,6 +138,7 @@ namespace NDiscovery {
                 const TMap<TActorId, TEvStateStorage::TBoardInfoEntry>& prevInfoEntries,
                 TMap<TActorId, TEvStateStorage::TBoardInfoEntry> newInfoEntries,
                 TSet<TString> services,
+                TString endpointId,
                 const THolder<TEvInterconnect::TEvNodeInfo>& nameserviceResponse) {
         TMap<TActorId, TEvStateStorage::TBoardInfoEntry> infoEntries;
         if (prevInfoEntries.empty()) {
@@ -170,6 +181,9 @@ namespace NDiscovery {
                 continue;
             }
 
+            if (!CheckEndpointId(endpointId, entry)) {
+                continue;
+            }
             if (entry.GetSsl()) {
                 AddEndpoint(cachedMessageSsl, statesSsl, entry);
             } else {
@@ -185,7 +199,6 @@ namespace NDiscovery {
                 cachedMessageSsl.set_self_location(location);
             }
         }
-
         return {SerializeResult(cachedMessage), SerializeResult(cachedMessageSsl), std::move(infoEntries)};
     }
 }
@@ -220,6 +233,7 @@ namespace NDiscoveryPrivate {
 
         THashMap<TString, TVector<TWaiter>> Requested;
         bool Scheduled = false;
+        TMaybe<TString> EndpointId;
 
         auto Request(const TString& database) {
             auto result = Requested.emplace(database, TVector<TWaiter>());
@@ -264,7 +278,8 @@ namespace NDiscoveryPrivate {
 
             currentCachedMessage = std::make_shared<NDiscovery::TCachedMessageData>(
                 NDiscovery::CreateCachedMessage(
-                    currentCachedMessage->InfoEntries, std::move(msg->Updates), {}, NameserviceResponse)
+                    currentCachedMessage->InfoEntries, std::move(msg->Updates),
+                    {}, EndpointId.GetOrElse({}), NameserviceResponse)
             );
 
             auto it = Requested.find(path);
@@ -278,7 +293,8 @@ namespace NDiscoveryPrivate {
             const auto& path = msg->Path;
 
             auto newCachedData = std::make_shared<NDiscovery::TCachedMessageData>(
-                NDiscovery::CreateCachedMessage({}, std::move(msg->InfoEntries), {}, NameserviceResponse)
+                NDiscovery::CreateCachedMessage({}, std::move(msg->InfoEntries),
+                {}, EndpointId.GetOrElse({}), NameserviceResponse)
             );
             newCachedData->Status = msg->Status;
 
@@ -357,6 +373,11 @@ namespace NDiscoveryPrivate {
         }
 
     public:
+        TDiscoveryCache() = default;
+        TDiscoveryCache(const TString& endpointId)
+            : EndpointId(endpointId)
+        {
+        }
         static constexpr NKikimrServices::TActivity::EType ActorActivityType() {
             return NKikimrServices::TActivity::DISCOVERY_CACHE_ACTOR;
         }
@@ -508,7 +529,7 @@ class TDiscoverer: public TActorBootstrapped<TDiscoverer> {
                 return true;
             default:
                 return true;
-        } 
+        }
     }
 
     void MaybeReply() {
@@ -606,8 +627,8 @@ IActor* CreateDiscoverer(
     return new TDiscoverer(f, database, replyTo, cacheId);
 }
 
-IActor* CreateDiscoveryCache() {
-    return new NDiscoveryPrivate::TDiscoveryCache();
+IActor* CreateDiscoveryCache(const TString& endpointId) {
+    return new NDiscoveryPrivate::TDiscoveryCache(endpointId);
 }
 
 }

ydb/core/discovery/discovery.h

@@ -59,6 +59,7 @@ namespace NDiscovery {
         const TMap<TActorId, TEvStateStorage::TBoardInfoEntry>&,
         TMap<TActorId, TEvStateStorage::TBoardInfoEntry>,
         TSet<TString>,
+        TString,
         const THolder<TEvInterconnect::TEvNodeInfo>&);
 }
 
@@ -74,6 +75,6 @@ IActor* CreateDiscoverer(
     const TActorId& cacheId);
 
 // Used to reduce number of requests to Board
-IActor* CreateDiscoveryCache();
+IActor* CreateDiscoveryCache(const TString& endpointId = {});
 
 }

ydb/core/driver_lib/run/config_parser.cpp

@@ -66,6 +66,7 @@ void TRunCommandConfigParser::SetupLastGetOptForConfigFiles(NLastGetopt::TOpts&
     opts.AddLongOption("grpc-file", "gRPC config file").OptionalArgument("PATH");
     opts.AddLongOption("grpc-port", "enable gRPC server on port").RequiredArgument("PORT");
     opts.AddLongOption("grpcs-port", "enable gRPC SSL server on port").RequiredArgument("PORT");
+    opts.AddLongOption("kafka-port", "enable kafka proxy server on port").OptionalArgument("PORT");
     opts.AddLongOption("grpc-public-host", "set public gRPC host for discovery").RequiredArgument("HOST");
     opts.AddLongOption("grpc-public-port", "set public gRPC port for discovery").RequiredArgument("PORT");
     opts.AddLongOption("grpcs-public-port", "set public gRPC SSL port for discovery").RequiredArgument("PORT");
@@ -164,6 +165,11 @@ void TRunCommandConfigParser::ParseConfigFiles(const NLastGetopt::TOptsParseResu
         conf.SetSslPort(FromString<ui16>(res.Get("grpcs-port")));
     }
 
+    if (res.Has("kafka-port")) {
+        auto& conf = *Config.AppConfig.MutableKafkaProxyConfig();
+        conf.SetListeningPort(FromString<ui16>(res.Get("kafka-port")));
+    }
+
     if (res.Has("grpc-public-host")) {
         auto& conf = *Config.AppConfig.MutableGRpcConfig();
         conf.SetPublicHost(res.Get("grpc-public-host"));