import {useEffect, useState} from "react"; import {Link} from "react-router-dom"; import {Loader2, Plus, RefreshCw, Trash2} from "lucide-react"; import {useMutation, useQuery, useQueryClient} from "@tanstack/react-query"; import {listCloudResources} from "@/api/cloudProxyClient"; import {getMskBootstrapBrokers, listMskClusters} from "@/api/mskClient"; import {createS3SinkConnector, deleteConnector, deleteCustomPlugin, describeConnector, listConnectors, listCustomPlugins, registerCustomPlugin, updateS3SinkConnector, type ConnectorDescription, type S3SinkConnectorInput} from "@/api/mskConnectClient"; const defaults: S3SinkConnectorInput = { connectorName: "", bootstrapServers: "", kafkaConnectVersion: "3.7.x", serviceExecutionRoleArn: "", customPluginArn: "", customPluginRevision: 1, workerCount: 1, mcuCount: 1, topics: "", keyConverter: "string", valueConverter: "json", s3Bucket: "", s3Region: "us-east-1", flushSize: "1", rotateIntervalMs: "60000", logGroup: "", }; export function MskConnectPanel({runtimeReachable}: {runtimeReachable: boolean}) { const queryClient = useQueryClient(); const [plugin, setPlugin] = useState({name: "", bucket: "", fileKey: "", objectVersion: ""}); const [form, setForm] = useState(defaults); const [selectedArn, setSelectedArn] = useState(""); const pluginsQuery = useQuery({queryKey: ["msk-connect-plugins"], queryFn: ({signal}) => listCustomPlugins(signal), enabled: runtimeReachable}); const clustersQuery = useQuery({queryKey: ["msk-connect-clusters"], queryFn: ({signal}) => listMskClusters(signal), enabled: runtimeReachable}); const connectorsQuery = useQuery({queryKey: ["msk-connect-connectors"], queryFn: ({signal}) => listConnectors(signal), enabled: runtimeReachable}); const bucketsQuery = useQuery({queryKey: ["cloud-resources", "aws", "storage", "msk-connect"], queryFn: ({signal}) => listCloudResources("aws", "storage", "", signal), enabled: runtimeReachable}); const connectorQuery = useQuery({queryKey: ["msk-connect-connector", selectedArn], queryFn: ({signal}) => describeConnector(selectedArn, signal), enabled: runtimeReachable && Boolean(selectedArn)}); const registerPlugin = useMutation({mutationFn: () => registerCustomPlugin(plugin), onSuccess: () => { setPlugin({name: "", bucket: "", fileKey: "", objectVersion: ""}); void queryClient.invalidateQueries({queryKey: ["msk-connect-plugins"]}); }}); const removePlugin = useMutation({mutationFn: (customPluginArn: string) => deleteCustomPlugin(customPluginArn), onSuccess: () => void queryClient.invalidateQueries({queryKey: ["msk-connect-plugins"]})}); const createConnector = useMutation({mutationFn: () => createS3SinkConnector(form), onSuccess: () => { setForm(defaults); void queryClient.invalidateQueries({queryKey: ["msk-connect-connectors"]}); }}); const updateConnector = useMutation({mutationFn: (connector: ConnectorDescription) => updateS3SinkConnector(selectedArn, {...form, currentVersion: connector.currentVersion ?? ""}), onSuccess: () => void queryClient.invalidateQueries({queryKey: ["msk-connect-connector", selectedArn]})}); const removeConnector = useMutation({mutationFn: () => deleteConnector(selectedArn), onSuccess: () => { setSelectedArn(""); void queryClient.invalidateQueries({queryKey: ["msk-connect-connectors"]}); }}); useEffect(() => { if (!selectedArn && connectorsQuery.data?.[0]?.connectorArn) setSelectedArn(connectorsQuery.data[0].connectorArn); }, [connectorsQuery.data, selectedArn]); useEffect(() => { const connector = connectorQuery.data; if (!connector) return; const configuration = connector.connectorConfiguration ?? {}; const capacity = connector.capacity?.provisionedCapacity; setForm((value) => ({...value, workerCount: capacity?.workerCount ?? value.workerCount, mcuCount: capacity?.mcuCount ?? value.mcuCount, topics: configuration.topics ?? value.topics, s3Bucket: configuration["s3.bucket.name"] ?? value.s3Bucket, s3Region: configuration["s3.region"] ?? value.s3Region, flushSize: configuration["flush.size"] ?? value.flushSize, rotateIntervalMs: configuration["rotate.interval.ms"] ?? value.rotateIntervalMs, keyConverter: configuration["key.converter"]?.includes("Json") ? "json" : "string", valueConverter: configuration["value.converter"]?.includes("String") ? "string" : "json"})); }, [connectorQuery.data]); const set = (key: K, value: S3SinkConnectorInput[K]) => setForm((current) => ({...current, [key]: value})); const error = registerPlugin.error ?? removePlugin.error ?? createConnector.error ?? updateConnector.error ?? removeConnector.error; const canCreate = form.connectorName.trim() && form.bootstrapServers.trim() && form.serviceExecutionRoleArn.trim() && form.customPluginArn && form.topics.trim() && form.s3Bucket; return

MSK Connect

Custom plugins

Register plugin ZIP archives from S3 using the AWS CreateCustomPlugin shape.

Browse S3
{ event.preventDefault(); if (plugin.name && plugin.bucket && plugin.fileKey) registerPlugin.mutate(); }}> bucket.name) ?? []} value={plugin.bucket} onChange={(bucket) => setPlugin({...plugin, bucket})}/> {pluginsQuery.data?.map((item) =>
{item.name}Revision {item.revision ?? 1} {item.customPluginState ? `(${item.customPluginState})` : ""}
)}

S3 Sink

Create connector

Uses the Confluent S3 Sink connector configuration, not an editable raw JSON payload.

{ event.preventDefault(); if (canCreate) createConnector.mutate(); }}> set("keyConverter", value)}/> set("valueConverter", value)}/> bucket.name) ?? []} value={form.s3Bucket} onChange={(value) => set("s3Bucket", value)}/>

Connectors

State and configuration

MSK Connect exposes connector state through DescribeConnector. It does not expose per-task status in its control-plane API.

{connectorQuery.data && } {selectedArn && connectorQuery.data &&
}{error instanceof Error &&

{error.message}

}
; } function BucketField({label, buckets, value, onChange}: {label: string; buckets: string[]; value: string; onChange: (value: string) => void}) { return ; } function ConverterField({label, value, onChange}: {label: string; value: "string" | "json"; onChange: (value: "string" | "json") => void}) { return ; } function ConnectorStatus({connector}: {connector: ConnectorDescription}) { const logGroup = connector.logDelivery?.workerLogDelivery?.cloudWatchLogs?.logGroup; return
{connector.connectorState ?? "Unknown"}{connector.stateDescription && {connector.stateDescription}}Task status: not available from the MSK Connect API{logGroup && Open {logGroup} in CloudWatch Logs}
; }