Fix for topic names

2 files changed, 44 insertions, 8 deletions

diff --git a/ydb/library/persqueue/topic_parser/topic_parser.cpp b/ydb/library/persqueue/topic_parser/topic_parser.cpp
index 72fa54f93a..88c94dcc76 100644
--- a/ydb/library/persqueue/topic_parser/topic_parser.cpp
+++ b/ydb/library/persqueue/topic_parser/topic_parser.cpp
@@ -18,6 +18,18 @@ bool IsPathPrefix(TStringBuf normPath, TStringBuf prefix) {
 }
 
 
+void SkipPathPrefix(TStringBuf& path, const TStringBuf& prefix) {
+    auto copy = path;
+    if (prefix.EndsWith('/')) {
+        path.SkipPrefix(prefix);
+    } else {
+        auto skip = path.SkipPrefix(prefix) && path.SkipPrefix("/");
+        if (!skip) {
+            path = copy;
+        }
+    }
+}
+
 namespace {
     TString FullPath(const TMaybe<TString>& database, const TString& path) {
         if (database.Defined() && !path.StartsWith(*database) && !path.Contains('\0')) {
@@ -193,21 +205,17 @@ void TDiscoveryConverter::BuildForFederation(const TStringBuf& databaseBuf, TStr
         if (IsPathPrefix(PQPrefix, databaseBuf)) {
             isRootDb = true;
             root = PQPrefix;
-            topicPath.SkipPrefix(PQPrefix);
-            topicPath.SkipPrefix("/");
+            SkipPathPrefix(topicPath, PQPrefix);
         } else {
-            topicPath.SkipPrefix(databaseBuf);
-            topicPath.SkipPrefix("/");
+            SkipPathPrefix(topicPath, databaseBuf);
         }
     } else if (IsPathPrefix(topicPath, PQPrefix)) {
         isRootDb = true;
-        topicPath.SkipPrefix(PQPrefix);
-        topicPath.SkipPrefix("/");
+        SkipPathPrefix(topicPath, PQPrefix);
         root = PQPrefix;
     }
     if (!isRootDb) {
-        topicPath.SkipPrefix(databaseBuf);
-        topicPath.SkipPrefix("/");
+        SkipPathPrefix(topicPath, databaseBuf);
         Database = databaseBuf;
     }
 
diff --git a/ydb/library/persqueue/topic_parser/ut/topic_names_converter_ut.cpp b/ydb/library/persqueue/topic_parser/ut/topic_names_converter_ut.cpp
index c8e3bbc804..118c81063b 100644
--- a/ydb/library/persqueue/topic_parser/ut/topic_names_converter_ut.cpp
+++ b/ydb/library/persqueue/topic_parser/ut/topic_names_converter_ut.cpp
@@ -140,6 +140,34 @@ Y_UNIT_TEST_SUITE(DiscoveryConverterTest) {
 //        UNIT_ASSERT_VALUES_EQUAL(wrapper.GetClientsideName(), "rt3.dc3--account@dir1@dir2--topic");
 
     }
+    Y_UNIT_TEST(CmWay) {
+        auto converterFactory = NPersQueue::TTopicNamesConverterFactory(
+                false, "/Root/PQ", "dc1", false
+        );
+        auto converter = converterFactory.MakeDiscoveryConverter(
+                "account/account-topic", {}, "dc1", "account"
+        );
+        UNIT_ASSERT_VALUES_EQUAL(converter->GetFullModernName(), "account-topic-mirrored-from-dc1");
+
+        converter = converterFactory.MakeDiscoveryConverter(
+                "account/topic", {}, "dc1", "account"
+        );
+        UNIT_ASSERT_VALUES_EQUAL(converter->GetFullModernName(), "topic-mirrored-from-dc1");
+
+        converterFactory = NPersQueue::TTopicNamesConverterFactory(
+                false, "/Root/PQ", "dc1", true
+        );
+        converter = converterFactory.MakeDiscoveryConverter(
+                "account/topic", {}, "dc1", "account"
+        );
+        UNIT_ASSERT_VALUES_EQUAL(converter->GetFullModernName(), "topic");
+
+        converter = converterFactory.MakeDiscoveryConverter(
+                "account/account-topic", {}, "dc1", "account"
+        );
+        UNIT_ASSERT_VALUES_EQUAL(converter->GetFullModernName(), "account-topic");
+
+        }
 
     Y_UNIT_TEST(FirstClass) {
         TConverterTestWrapper wrapper(true, "", "");