fix(kafka): support cluster address separators
Co-authored-by: zipg <4047349+zipg@users.noreply.github.com>
This commit is contained in:
1 parent
51693bf296
commit
a212ccbfa7
3 files changed
+60
-28
No files matched your search
@@ -39,6 +39,7 @@ import { prestoSqlBuiltinDriverPaths } from "@/lib/database/prestoSqlBuiltinDriv
|
|||||||
import { SQLITE_DATABASE_FILE_EXTENSIONS } from "@/lib/database/databaseFileDetection";
|
import { SQLITE_DATABASE_FILE_EXTENSIONS } from "@/lib/database/databaseFileDetection";
|
||||||
import { connectionAttemptOriginalErrorMessage, connectionAttemptTimeoutMessage, connectionAttemptTimeoutMs } from "@/lib/connection/connectionAttemptTimeout";
|
import { connectionAttemptOriginalErrorMessage, connectionAttemptTimeoutMessage, connectionAttemptTimeoutMs } from "@/lib/connection/connectionAttemptTimeout";
|
||||||
import { appendConnectionErrorHints } from "@/lib/connection/connectionErrorHints";
|
import { appendConnectionErrorHints } from "@/lib/connection/connectionErrorHints";
|
||||||
|
import { normalizeKafkaBootstrapServers } from "@/lib/connection/kafkaBootstrapServers";
|
||||||
import { driverInstallProgressPercent, type DriverInstallProgress } from "@/lib/connection/driverInstallProgressUi";
|
import { driverInstallProgressPercent, type DriverInstallProgress } from "@/lib/connection/driverInstallProgressUi";
|
||||||
import { ArrowLeft, ArrowDown, ArrowUp, CheckSquare, ChevronRight, CircleHelp, Copy, ExternalLink, FilePlus2, FolderOpen, GripVertical, Grid3X3, KeyRound, Link2, List, ListFilter, Loader2, Pencil, Pipette, Plus, Search, ShieldCheck, Square, Trash2 } from "@lucide/vue";
|
import { ArrowLeft, ArrowDown, ArrowUp, CheckSquare, ChevronRight, CircleHelp, Copy, ExternalLink, FilePlus2, FolderOpen, GripVertical, Grid3X3, KeyRound, Link2, List, ListFilter, Loader2, Pencil, Pipette, Plus, Search, ShieldCheck, Square, Trash2 } from "@lucide/vue";
|
||||||
import { buildDraftVisibleDatabasesConnectionId, connectionCanChooseVisibleDatabases, initialVisibleDatabaseSelection, visibleDatabaseSelectionIsStale } from "@/lib/connection/connectionVisibleDatabases";
|
import { buildDraftVisibleDatabasesConnectionId, connectionCanChooseVisibleDatabases, initialVisibleDatabaseSelection, visibleDatabaseSelectionIsStale } from "@/lib/connection/connectionVisibleDatabases";
|
||||||
@@ -795,32 +796,6 @@ function requireMqField(value: string, message: string): string {
|
|||||||
return trimmed;
|
return trimmed;
|
||||||
}
|
}
|
||||||
|
|
||||||
function normalizeMqKafkaBootstrapServer(server: string): string {
|
|
||||||
if (server.includes("://")) {
|
|
||||||
throw new Error("Kafka bootstrap servers must be host:port values without a URL scheme");
|
|
||||||
}
|
|
||||||
let parsed: URL;
|
|
||||||
try {
|
|
||||||
parsed = new URL(`kafka://${server}`);
|
|
||||||
} catch {
|
|
||||||
throw new Error("Kafka bootstrap servers are invalid");
|
|
||||||
}
|
|
||||||
if (!parsed.hostname || parsed.username || parsed.password || parsed.search || parsed.hash || (parsed.pathname && parsed.pathname !== "/")) {
|
|
||||||
throw new Error("Kafka bootstrap servers are invalid");
|
|
||||||
}
|
|
||||||
return server;
|
|
||||||
}
|
|
||||||
|
|
||||||
function normalizeMqKafkaBootstrapServers(value: string): string {
|
|
||||||
const servers = requireMqField(value, "Kafka bootstrap servers are required")
|
|
||||||
.split(",")
|
|
||||||
.map((server) => server.trim())
|
|
||||||
.filter(Boolean)
|
|
||||||
.map(normalizeMqKafkaBootstrapServer);
|
|
||||||
if (!servers.length) throw new Error("Kafka bootstrap servers are required");
|
|
||||||
return servers.join(",");
|
|
||||||
}
|
|
||||||
|
|
||||||
function buildMqAuth(): MqAuth {
|
function buildMqAuth(): MqAuth {
|
||||||
switch (mqAuthKind.value) {
|
switch (mqAuthKind.value) {
|
||||||
case "token":
|
case "token":
|
||||||
@@ -862,7 +837,7 @@ function buildMqTokenSigning() {
|
|||||||
function buildMqAdminConfig(): MqAdminConfig {
|
function buildMqAdminConfig(): MqAdminConfig {
|
||||||
const systemKind = mqSystemKind.value;
|
const systemKind = mqSystemKind.value;
|
||||||
if (systemKind === "kafka") {
|
if (systemKind === "kafka") {
|
||||||
const bootstrapServers = normalizeMqKafkaBootstrapServers(mqKafkaBootstrapServers.value);
|
const bootstrapServers = normalizeKafkaBootstrapServers(mqKafkaBootstrapServers.value);
|
||||||
const extra: Record<string, string> = { bootstrapServers };
|
const extra: Record<string, string> = { bootstrapServers };
|
||||||
const securityProtocol = mqKafkaSecurityProtocol.value === MQ_KAFKA_SECURITY_PROTOCOL_AUTO ? "" : mqKafkaSecurityProtocol.value.trim();
|
const securityProtocol = mqKafkaSecurityProtocol.value === MQ_KAFKA_SECURITY_PROTOCOL_AUTO ? "" : mqKafkaSecurityProtocol.value.trim();
|
||||||
const saslMechanism = mqKafkaSaslMechanism.value.trim();
|
const saslMechanism = mqKafkaSaslMechanism.value.trim();
|
||||||
@@ -1107,7 +1082,7 @@ function applyMqAdminUrl(config: LegacyConnectionConfig, adminUrl: string) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
function applyMqKafkaBootstrapServers(config: LegacyConnectionConfig, bootstrapServers: string, securityProtocol?: string) {
|
function applyMqKafkaBootstrapServers(config: LegacyConnectionConfig, bootstrapServers: string, securityProtocol?: string) {
|
||||||
const first = normalizeMqKafkaBootstrapServers(bootstrapServers).split(",")[0];
|
const first = normalizeKafkaBootstrapServers(bootstrapServers).split(",")[0];
|
||||||
if (!first) throw new Error("Kafka bootstrap servers are required");
|
if (!first) throw new Error("Kafka bootstrap servers are required");
|
||||||
let parsed: URL;
|
let parsed: URL;
|
||||||
try {
|
try {
|
||||||
|
|||||||
@@ -0,0 +1,24 @@
|
|||||||
|
import { describe, expect, it } from "vitest";
|
||||||
|
import { normalizeKafkaBootstrapServers } from "@/lib/connection/kafkaBootstrapServers";
|
||||||
|
|
||||||
|
describe("Kafka bootstrap servers", () => {
|
||||||
|
it("keeps comma-separated bootstrap servers", () => {
|
||||||
|
expect(normalizeKafkaBootstrapServers("broker1:9092, broker2:9092")).toBe("broker1:9092,broker2:9092");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("normalizes common cluster address separators to commas", () => {
|
||||||
|
expect(normalizeKafkaBootstrapServers("broker1:9092;broker2:9092,broker3:9092\nbroker4:9092 broker5:9092")).toBe("broker1:9092,broker2:9092,broker3:9092,broker4:9092,broker5:9092");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("keeps IPv6 bootstrap servers", () => {
|
||||||
|
expect(normalizeKafkaBootstrapServers("[::1]:9092;[2001:db8::1]:9092")).toBe("[::1]:9092,[2001:db8::1]:9092");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("rejects bootstrap servers with URL schemes", () => {
|
||||||
|
expect(() => normalizeKafkaBootstrapServers("PLAINTEXT://broker1:9092,broker2:9092")).toThrow("Kafka bootstrap servers must be host:port values without a URL scheme");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("rejects invalid bootstrap server values", () => {
|
||||||
|
expect(() => normalizeKafkaBootstrapServers("broker1:9092/path,broker2:9092")).toThrow("Kafka bootstrap servers are invalid");
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -0,0 +1,33 @@
|
|||||||
|
const KAFKA_BOOTSTRAP_SERVER_SEPARATOR = /[\s,;,;]+/u;
|
||||||
|
|
||||||
|
function requireKafkaBootstrapServers(value: string): string {
|
||||||
|
const trimmed = value.trim();
|
||||||
|
if (!trimmed) throw new Error("Kafka bootstrap servers are required");
|
||||||
|
return trimmed;
|
||||||
|
}
|
||||||
|
|
||||||
|
function normalizeKafkaBootstrapServer(server: string): string {
|
||||||
|
if (server.includes("://")) {
|
||||||
|
throw new Error("Kafka bootstrap servers must be host:port values without a URL scheme");
|
||||||
|
}
|
||||||
|
let parsed: URL;
|
||||||
|
try {
|
||||||
|
parsed = new URL(`kafka://${server}`);
|
||||||
|
} catch {
|
||||||
|
throw new Error("Kafka bootstrap servers are invalid");
|
||||||
|
}
|
||||||
|
if (!parsed.hostname || parsed.username || parsed.password || parsed.search || parsed.hash || (parsed.pathname && parsed.pathname !== "/")) {
|
||||||
|
throw new Error("Kafka bootstrap servers are invalid");
|
||||||
|
}
|
||||||
|
return server;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function normalizeKafkaBootstrapServers(value: string): string {
|
||||||
|
const servers = requireKafkaBootstrapServers(value)
|
||||||
|
.split(KAFKA_BOOTSTRAP_SERVER_SEPARATOR)
|
||||||
|
.map((server) => server.trim())
|
||||||
|
.filter(Boolean)
|
||||||
|
.map(normalizeKafkaBootstrapServer);
|
||||||
|
if (!servers.length) throw new Error("Kafka bootstrap servers are required");
|
||||||
|
return servers.join(",");
|
||||||
|
}
|
||||||
Reference in new issue
Block a user