Support dekaf materializations

src/components/capture/useCaptureDraftUpdate.ts

@@ -23,6 +23,7 @@ import {
     useEndpointConfigStore_setEncryptedEndpointConfig,
     useEndpointConfigStore_setPreviousEndpointConfig,
 } from 'stores/EndpointConfig/hooks';
+import { isDekafConnector } from 'utils/misc-utils';
 
 import { modifyExistingCaptureDraftSpec } from 'utils/workflow-utils';
 
@@ -38,8 +39,10 @@ function useDiscoverDraftUpdate(options?: {
     const persistedDraftId = useEditorStore_persistedDraftId();
     const setDraftId = useEditorStore_setId();
 
-    const imagePath = useDetailsFormStore(
-        (state) => state.details.data.connectorImage.imagePath
+    const imagePath = useDetailsFormStore((state) =>
+        isDekafConnector(state.details.data.connectorImage)
+            ? undefined
+            : state.details.data.connectorImage.imagePath
     );
 
     const setEncryptedEndpointConfig =
@@ -67,10 +70,13 @@ function useDiscoverDraftUpdate(options?: {
             //      Doing a rediscovery with no new changes
 
             if (
-                (!draftId && persistedDraftId && !options?.initiateDiscovery) ||
-                (persistedDraftId &&
-                    !options?.initiateDiscovery &&
-                    !options?.initiateRediscovery)
+                imagePath &&
+                ((!draftId &&
+                    persistedDraftId &&
+                    !options?.initiateDiscovery) ||
+                    (persistedDraftId &&
+                        !options?.initiateDiscovery &&
+                        !options?.initiateRediscovery))
             ) {
                 const existingDraftSpecResponse =
                     await getDraftSpecsByCatalogName(

src/components/materialization/useGenerateCatalog.ts

@@ -28,22 +28,25 @@ import {
 } from 'stores/Binding/hooks';
 import { useDetailsFormStore } from 'stores/DetailsForm/Store';
 import {
-    useEndpointConfig_serverUpdateRequired,
     useEndpointConfigStore_encryptedEndpointConfig_data,
     useEndpointConfigStore_endpointConfig_data,
     useEndpointConfigStore_endpointSchema,
     useEndpointConfigStore_errorsExist,
     useEndpointConfigStore_setEncryptedEndpointConfig,
     useEndpointConfigStore_setPreviousEndpointConfig,
+    useEndpointConfig_serverUpdateRequired,
 } from 'stores/EndpointConfig/hooks';
 import {
     useFormStateStore_setFormState,
     useFormStateStore_updateStatus,
 } from 'stores/FormState/hooks';
 import { FormStatus } from 'stores/FormState/types';
 import { useSourceCaptureStore } from 'stores/SourceCapture/Store';
+import { DekafConfig } from 'types';
+import { isDekafConnector } from 'utils/misc-utils';
 import { encryptEndpointConfig } from 'utils/sops-utils';
 import { generateTaskSpec } from 'utils/workflow-utils';
+import { ConnectorConfig } from '../../../deps/flow/flow';
 
 const ENTITY_TYPE = 'materialization';
 
@@ -61,8 +64,17 @@ function useGenerateCatalog() {
     const imageConnectorId = useDetailsFormStore(
         (state) => state.details.data.connectorImage.connectorId
     );
-    const imagePath = useDetailsFormStore(
-        (state) => state.details.data.connectorImage.imagePath
+    const endpointConfig: ConnectorConfig | DekafConfig = useDetailsFormStore(
+        (state) =>
+            isDekafConnector(state.details.data.connectorImage)
+                ? {
+                      config: {},
+                      variant: state.details.data.connectorImage.variant,
+                  }
+                : {
+                      config: {},
+                      image: state.details.data.connectorImage.imagePath,
+                  }
     );
     const setDraftedEntityName = useDetailsFormStore(
         (state) => state.setDraftedEntityName
@@ -207,12 +219,11 @@ function useGenerateCatalog() {
                     { overrideJsonFormDefaults: true }
                 );
 
+                endpointConfig.config = encryptedEndpointConfig.data;
+
                 const draftSpec = generateTaskSpec(
                     ENTITY_TYPE,
-                    {
-                        image: imagePath,
-                        config: encryptedEndpointConfig.data,
-                    },
+                    endpointConfig,
                     resourceConfigs,
                     resourceConfigServerUpdateRequired,
                     bindings,
@@ -274,8 +285,13 @@ function useGenerateCatalog() {
                 // Update all the store state
                 setCatalogName(processedEntityName);
                 setEncryptedEndpointConfig({
-                    data: draftSpecsResponse.data[0].spec.endpoint.connector
-                        .config,
+                    data: Object.hasOwn(
+                        draftSpecsResponse.data[0].spec.endpoint,
+                        'dekaf'
+                    )
+                        ? draftSpecsResponse.data[0].spec.endpoint.dekaf.config
+                        : draftSpecsResponse.data[0].spec.endpoint.connector
+                              .config,
                 });
                 setPreviousEndpointConfig({ data: endpointConfigData });
                 setDraftId(evaluatedDraftId);
@@ -303,14 +319,14 @@ function useGenerateCatalog() {
             bindings,
             callFailed,
             detailsFormsErrorsExist,
+            endpointConfig,
             endpointConfigData,
             endpointConfigErrorsExist,
             endpointSchema,
             fullSourceConfigs,
             fullSourceErrorsExist,
             imageConnectorId,
             imageConnectorTagId,
-            imagePath,
             persistedDraftId,
             prefillBindingDependentState,
             processedEntityName,

src/components/shared/Entity/DetailsForm/Form.tsx

@@ -5,37 +5,19 @@ import { useEditorStore_isSaving } from 'components/editor/Store/hooks';
 import AlertBox from 'components/shared/AlertBox';
 import { Props } from 'components/shared/Entity/DetailsForm/types';
 import { CONNECTOR_IMAGE_SCOPE } from 'forms/renderers/Connectors';
-import { ConnectorWithTagDetailQuery } from 'hooks/connectors/shared';
 import { FormattedMessage, useIntl } from 'react-intl';
 import defaultRenderers from 'services/jsonforms/defaultRenderers';
 import { defaultOptions, showValidation } from 'services/jsonforms/shared';
 import { useDetailsFormStore } from 'stores/DetailsForm/Store';
-import { Details } from 'stores/DetailsForm/types';
 import {
     useFormStateStore_displayValidation,
     useFormStateStore_isActive,
     useFormStateStore_messagePrefix,
 } from 'stores/FormState/hooks';
-import { evaluateConnectorVersions } from 'utils/workflow-utils';
 import useFormFields from './useFormFields';
 
 export const CONFIG_EDITOR_ID = 'endpointConfigEditor';
 
-export const getConnectorImageDetails = (
-    connector: ConnectorWithTagDetailQuery,
-    options?: { connectorId: string; existingImageTag: string }
-): Details['data']['connectorImage'] => {
-    const connectorTag = evaluateConnectorVersions(connector, options);
-
-    return {
-        connectorId: connector.id,
-        id: connectorTag.id,
-        imageName: connector.image_name,
-        imagePath: `${connector.image_name}${connectorTag.image_tag}`,
-        iconPath: connector.image,
-    };
-};
-
 function DetailsFormForm({ connectorTags, entityType, readOnly }: Props) {
     const intl = useIntl();
 

src/components/shared/Entity/DetailsForm/useConnectorField.ts

@@ -6,9 +6,9 @@ import useGlobalSearchParams, {
 } from 'hooks/searchParams/useGlobalSearchParams';
 import { useCallback, useEffect, useMemo } from 'react';
 import { useIntl } from 'react-intl';
-import { useDetailsFormStore } from 'stores/DetailsForm/Store';
 import { useDetailsForm_changed_connectorId } from 'stores/DetailsForm/hooks';
-import { Details } from 'stores/DetailsForm/types';
+import { useDetailsFormStore } from 'stores/DetailsForm/Store';
+import { ConnectorMetadata, Details } from 'stores/DetailsForm/types';
 import { EntityWithCreateWorkflow } from 'types';
 import { hasLength } from 'utils/misc-utils';
 import {
@@ -17,7 +17,35 @@ import {
 } from 'utils/workflow-utils';
 import { MAC_ADDR_RE } from 'validation';
 import useEntityCreateNavigate from '../hooks/useEntityCreateNavigate';
-import { getConnectorImageDetails } from './Form';
+
+// TODO (optimization): Consider deriving ConnectorWithTagDetailQuery and ConnectorsQuery_DetailsForm
+//   from the same base interface so this function can be used in getConnectorImage.
+const getConnectorImageDetails = (
+    connector: ConnectorWithTagDetailQuery,
+    options?: { connectorId: string; existingImageTag: string }
+): Details['data']['connectorImage'] => {
+    const connectorTag = evaluateConnectorVersions(connector, options);
+
+    const { image_name } = connector;
+
+    const connectorMetadata: ConnectorMetadata = {
+        connectorId: connector.id,
+        iconPath: connector.image,
+        id: connectorTag.id,
+        imageName: image_name,
+        imageTag: connectorTag.image_tag,
+    };
+
+    return connectorTag.image_tag.includes(':dekaf')
+        ? {
+              ...connectorMetadata,
+              variant: `${image_name}${connectorTag.image_tag}`,
+          }
+        : {
+              ...connectorMetadata,
+              imagePath: `${image_name}${connectorTag.image_tag}`,
+          };
+};
 
 export default function useConnectorField(
     connectorTags: ConnectorWithTagDetailQuery[],
@@ -33,8 +61,8 @@ export default function useConnectorField(
     const originalConnectorImage = useDetailsFormStore(
         (state) => state.details.data.connectorImage
     );
-    const connectorImagePath = useDetailsFormStore(
-        (state) => state.details.data.connectorImage.imagePath
+    const connectorImageTag = useDetailsFormStore(
+        (state) => state.details.data.connectorImage.imageTag
     );
     const connectorIdChanged = useDetailsForm_changed_connectorId();
     const setDetails_connector = useDetailsFormStore(
@@ -66,21 +94,17 @@ export default function useConnectorField(
         | undefined = useMemo(() => {
         // This is rare but can happen so being safe.
         // If you remove the connector id from the create URL
-        if (!connectorImagePath) {
+        if (!connectorImageTag) {
             return undefined;
         }
 
-        const imageTagStartIndex = connectorImagePath.indexOf(':');
-        return isEdit && hasLength(connectorId) && imageTagStartIndex > 0
+        return isEdit && hasLength(connectorId)
             ? {
                   connectorId,
-                  existingImageTag: connectorImagePath.substring(
-                      imageTagStartIndex,
-                      connectorImagePath.length
-                  ),
+                  existingImageTag: connectorImageTag,
               }
             : undefined;
-    }, [connectorId, connectorImagePath, isEdit]);
+    }, [connectorId, connectorImageTag, isEdit]);
 
     const connectorsOneOf = useMemo(() => {
         const response = [] as { title: string; const: Object }[];

src/components/shared/Entity/EndpointConfig/index.tsx

@@ -253,6 +253,16 @@ function EndpointConfig({
                         </Box>
                     ) : null}
 
+                    {canBeEmpty ? (
+                        <Box sx={{ mb: 3 }}>
+                            <AlertBox severity="info" short>
+                                {intl.formatMessage({
+                                    id: 'workflows.alert.endpointConfigEmpty',
+                                })}
+                            </AlertBox>
+                        </Box>
+                    ) : null}
+
                     <EndpointConfigForm readOnly={readOnly} />
                 </ErrorBoundryWrapper>
             </WrapperWithHeader>

src/lang/en-US/Workflows.ts

@@ -193,6 +193,8 @@ export const Workflows: Record<string, string> = {
     'workflows.dataPlane.description': `Choose the data plane you would like to use.`,
     'workflows.dataPlane.label': `Data Plane`,
 
+    'workflows.alert.endpointConfigEmpty': `This endpoint requires configuration of the external system only.`,
+
     // Field Selection
     'fieldSelection.header': `Field Selection`,
     'fieldSelection.table.label': `Field Selection Editor`,

src/stores/DetailsForm/Store.ts

@@ -7,6 +7,7 @@ import { isEmpty } from 'lodash';
 import { logRocketEvent } from 'services/shared';
 import { CustomEvents } from 'services/types';
 import {
+    ConnectorMetadata,
     DataPlaneOption,
     Details,
     DetailsFormState,
@@ -55,13 +56,23 @@ const getConnectorImage = async (
 
         const connectorTag = evaluateConnectorVersions(connector, options);
 
-        return {
+        const connectorMetadata: ConnectorMetadata = {
             connectorId,
+            iconPath: logo_url,
             id: connectorTag.id,
             imageName: image_name,
-            imagePath: `${image_name}${connectorTag.image_tag}`,
-            iconPath: logo_url,
+            imageTag: connectorTag.image_tag,
         };
+
+        return connectorTag.image_tag.includes(':dekaf')
+            ? {
+                  ...connectorMetadata,
+                  variant: `${image_name}${connectorTag.image_tag}`,
+              }
+            : {
+                  ...connectorMetadata,
+                  imagePath: `${image_name}${connectorTag.image_tag}`,
+              };
     }
 
     return null;
@@ -103,6 +114,7 @@ const initialDetails: Details = {
             iconPath: '',
             imageName: '',
             imagePath: '',
+            imageTag: '',
         },
         entityName: '',
     },
@@ -413,6 +425,7 @@ export const getInitialState = (
                     iconPath: '',
                     imageName: '',
                     imagePath: '',
+                    imageTag: '',
                     connectorId,
                 });
                 get().setHydrationErrorsExist(true);

src/stores/DetailsForm/types.ts

@@ -18,15 +18,25 @@ export interface DataPlaneOption {
     scope: 'public' | 'private';
 }
 
+export interface ConnectorMetadata {
+    connectorId: string;
+    iconPath: string;
+    id: string;
+    imageName: string;
+    imageTag: string;
+}
+
+export interface StandardConnectorMetadata extends ConnectorMetadata {
+    imagePath: string;
+}
+
+export interface DekafConnectorMetadata extends ConnectorMetadata {
+    variant: string;
+}
+
 export interface Details extends Pick<JsonFormsCore, 'data' | 'errors'> {
     data: {
-        connectorImage: {
-            id: string;
-            iconPath: string;
-            imageName: string;
-            imagePath: string;
-            connectorId: string;
-        };
+        connectorImage: StandardConnectorMetadata | DekafConnectorMetadata;
         entityName: string;
         dataPlane?: DataPlaneOption;
     };