Stopping transfer and async replication if connection_string is bad (#18943)

Author: nshestakov

ydb/core/transfer/transfer_writer.cpp

@@ -781,7 +781,7 @@ class TTransferWriter
     }
 
     void Handle(TEvTxUserProxy::TEvUploadRowsResponse::TPtr& ev) {
-        LOG_E("Handle TEvTxUserProxy::TEvUploadRowsResponse"
+        LOG_D("Handle TEvTxUserProxy::TEvUploadRowsResponse"
             << ": worker# " << Worker
             << " status# " << ev->Get()->Status
             << " issues# " << ev->Get()->Issues.ToOneLineString());

ydb/core/transfer/ut/functional/transfer_ut.cpp

@@ -22,6 +22,83 @@ Y_UNIT_TEST_SUITE(Transfer)
         testCase.DropTopic();
     }
 
+    Y_UNIT_TEST(ConnectionString_BadChar)
+    {
+        MainTestCase testCase;
+
+        testCase.CreateTable(R"(
+                CREATE TABLE `%s` (
+                    Key Uint64 NOT NULL,
+                    Message Utf8,
+                    PRIMARY KEY (Key)
+                )  WITH (
+                    STORE = COLUMN
+                );
+            )");
+        testCase.CreateTopic(1);
+        testCase.ExecuteDDL(Sprintf(R"(
+                $l = ($x) -> {
+                    return [
+                        <|
+                            Key: 1,
+                            Message:CAST("Message-1" AS Utf8)
+                        |>
+                    ];
+                };
+
+                CREATE TRANSFER %s
+                FROM %s TO %s USING $l
+                WITH (
+                    CONNECTION_STRING = "grp§c://localhost:2135"
+                )
+            )", testCase.TransferName.data(), testCase.TopicName.data(), testCase.TableName.data()));
+
+        testCase.CheckTransferStateError("DNS resolution failed for grp§c://localhost:2135");
+
+        testCase.DropTransfer();
+        testCase.DropTable();
+        testCase.DropTopic();
+    }
+
+    Y_UNIT_TEST(ConnectionString_BadDNSName)
+    {
+        MainTestCase testCase;
+
+        testCase.CreateTable(R"(
+                CREATE TABLE `%s` (
+                    Key Uint64 NOT NULL,
+                    Message Utf8,
+                    PRIMARY KEY (Key)
+                )  WITH (
+                    STORE = COLUMN
+                );
+            )");
+        testCase.CreateTopic(1);
+        testCase.ExecuteDDL(Sprintf(R"(
+                $l = ($x) -> {
+                    return [
+                        <|
+                            Key: 1,
+                            Message:CAST("Message-1" AS Utf8)
+                        |>
+                    ];
+                };
+
+                CREATE TRANSFER %s
+                FROM %s TO %s USING $l
+                WITH (
+                    CONNECTION_STRING = "grpc://domain-not-exists-localhost.com.moc:2135"
+                )
+            )", testCase.TransferName.data(), testCase.TopicName.data(), testCase.TableName.data()));
+
+        testCase.CheckTransferStateError("Grpc error response on endpoint domain-not-exists-localhost.com.moc:2135");
+
+        testCase.DropTransfer();
+        testCase.DropTable();
+        testCase.DropTopic();
+    }
+
+
     Y_UNIT_TEST(Create_WithPermission)
     {
         auto id = RandomNumber<ui16>();

ydb/core/tx/replication/controller/util.h

@@ -41,6 +41,13 @@ inline bool IsRetryableError(const NYdb::TStatus status, const TVector<NYdb::ESt
     case NYdb::EStatus::CLIENT_UNAUTHENTICATED:
     case NYdb::EStatus::CLIENT_CALL_UNIMPLEMENTED:
         return false;
+    case NYdb::EStatus::TRANSPORT_UNAVAILABLE:
+        for (const auto& issue : status.GetIssues()) {
+            if (issue.GetMessage().contains("Misformatted domain name") || issue.GetMessage().contains("Domain name not found")) {
+                return false;
+            }
+        }
+        return true;
     default:
         return status.IsTransportError() || Find(retryable, status.GetStatus()) != retryable.end();
     }