feat: pause reconciliation

ui/api/kafka/actions.ts

@@ -365,3 +365,36 @@ export async function getKafkaTopicMetrics(
     };
   }
 }
+
+export async function updateKafkaCluster(
+  clusterId: string,
+  reconciliationPaused?: boolean,
+): Promise<boolean> {
+  const url = `${process.env.BACKEND_URL}/api/kafkas/${clusterId}`;
+  const body = {
+    data: {
+      type: "kafkas",
+      id: clusterId,
+      meta: {
+        reconciliationPaused: reconciliationPaused,
+      },
+      attributes: {},
+    },
+  };
+
+  try {
+    const res = await fetch(url, {
+      headers: await getHeaders(),
+      method: "PATCH",
+      body: JSON.stringify(body),
+    });
+
+    if (res.status === 200) {
+      return true;
+    } else {
+      return false;
+    }
+  } catch (e) {
+    return false;
+  }
+}

ui/api/kafka/schema.ts

@@ -41,6 +41,11 @@ export type ClusterList = z.infer<typeof ClusterListSchema>;
 const ClusterDetailSchema = z.object({
   id: z.string(),
   type: z.literal("kafkas"),
+  meta: z
+    .object({
+      reconciliationPaused: z.boolean().optional(),
+    })
+    .optional(),
   attributes: z.object({
     name: z.string(),
     namespace: z.string().nullable().optional(),

ui/app/[locale]/(authorized)/kafka/[kafkaId]/@header/overview/ConnectButton.tsx

@@ -1,15 +1,65 @@
 "use client";
 
+import { updateKafkaCluster } from "@/api/kafka/actions";
 import { useOpenClusterConnectionPanel } from "@/components/ClusterDrawerContext";
-import { Button } from "@/libs/patternfly/react-core";
+import { ReconciliationModal } from "@/components/ClusterOverview/ReconciliationModal";
+import { useReconciliationContext } from "@/components/ReconciliationContext";
+import { Button, Flex, FlexItem } from "@/libs/patternfly/react-core";
 import { useTranslations } from "next-intl";
+import { useState } from "react";
 
 export function ConnectButton({ clusterId }: { clusterId: string }) {
   const t = useTranslations();
   const open = useOpenClusterConnectionPanel();
+
+  const [isModalOpen, setIsModalOpen] = useState<boolean>(false);
+
+  const { isReconciliationPaused, setReconciliationPaused } =
+    useReconciliationContext();
+
+  const onClickUpdate = async (pausedState: boolean) => {
+    try {
+      const success = await updateKafkaCluster(clusterId, pausedState);
+
+      if (success) {
+        setReconciliationPaused(pausedState);
+        setIsModalOpen(false);
+      }
+    } catch (e: unknown) {
+      console.log("Unknown error occurred");
+    }
+  };
+
   return (
-    <Button onClick={() => open(clusterId)}>
-      {t("ConnectButton.cluster_connection_details")}
-    </Button>
+    <>
+      <Flex>
+        <FlexItem>
+          <Button
+            variant="secondary"
+            onClick={
+              isReconciliationPaused
+                ? () => onClickUpdate(false)
+                : () => setIsModalOpen(true)
+            }
+          >
+            {isReconciliationPaused
+              ? t("reconciliation.resume_reconciliation")
+              : t("reconciliation.pause_reconciliation_button")}
+          </Button>
+        </FlexItem>
+        <FlexItem>
+          <Button onClick={() => open(clusterId)}>
+            {t("ConnectButton.cluster_connection_details")}
+          </Button>
+        </FlexItem>
+      </Flex>
+      {isModalOpen && (
+        <ReconciliationModal
+          isModalOpen={isModalOpen}
+          onClickClose={() => setIsModalOpen(false)}
+          onClickPauseReconciliation={() => onClickUpdate(true)}
+        />
+      )}
+    </>
   );
 }

ui/app/[locale]/(authorized)/kafka/[kafkaId]/@header/overview/page.tsx

@@ -29,6 +29,7 @@ async function ConnectedHeader({
   if (!cluster) {
     notFound();
   }
+
   return <OverviewHeader params={{ kafkaId }} />;
 }
 

ui/app/[locale]/(authorized)/kafka/[kafkaId]/layout.tsx

@@ -11,6 +11,8 @@ import { getServerSession } from "next-auth";
 import { useTranslations } from "next-intl";
 import { PropsWithChildren, ReactNode, Suspense } from "react";
 import { KafkaParams } from "./kafka.params";
+import { notFound } from "next/navigation";
+import { getKafkaCluster } from "@/api/kafka/actions";
 
 export default async function AsyncLayout({
   children,
@@ -26,6 +28,13 @@ export default async function AsyncLayout({
 }>) {
   const authOptions = await getAuthOptions();
   const session = await getServerSession(authOptions);
+
+  const cluster = await getKafkaCluster(kafkaId);
+
+  if (!cluster) {
+    notFound();
+  }
+
   return (
     <Layout
       username={(session?.user?.name || session?.user?.email) ?? "User"}
@@ -58,6 +67,7 @@ function Layout({
     <AppLayoutProvider>
       <AppLayout
         username={username}
+        kafkaId={kafkaId}
         sidebar={<ClusterLinks kafkaId={kafkaId} />}
       >
         <PageGroup stickyOnBreakpoint={{ default: "top" }}>

ui/app/[locale]/(authorized)/kafka/[kafkaId]/overview/ConnectedClusterCard.tsx

@@ -14,30 +14,31 @@ export async function ConnectedClusterCard({
     return (
       <ClusterCard
         isLoading={false}
-        status={res?.cluster.attributes.status || "n/a"}
+        status={res?.cluster.attributes.status ?? "n/a"}
         messages={[]}
-        name={res?.cluster.attributes.name || "n/a"}
+        name={res?.cluster.attributes.name ?? "n/a"}
         consumerGroups={undefined}
         brokersOnline={undefined}
         brokersTotal={undefined}
-        kafkaVersion={res?.cluster.attributes.kafkaVersion || "n/a"}
+        kafkaVersion={res?.cluster.attributes.kafkaVersion ?? "n/a"}
+        kafkaId={res?.cluster.id}
       />
     );
   }
   const groupCount = await consumerGroups.then(
     (grpResp) => grpResp?.meta.page.total ?? 0,
   );
-  const brokersTotal = Object.keys(res?.kpis.broker_state || {}).length;
+  const brokersTotal = Object.keys(res?.kpis.broker_state ?? {}).length;
   const brokersOnline =
-    Object.values(res?.kpis.broker_state || {}).filter((s) => s === 3).length ||
+    Object.values(res?.kpis.broker_state ?? {}).filter((s) => s === 3).length ||
     0;
   const messages = res?.cluster.attributes.conditions
     ?.filter((c) => "Ready" !== c.type)
     .map((c) => ({
       variant:
         c.type === "Error" ? "danger" : ("warning" as "danger" | "warning"),
       subject: {
-        type: "cluster" as "cluster" | "broker" | "topic",
+        type: c.type!,
         name: res?.cluster.attributes.name ?? "",
         id: res?.cluster.id ?? "",
       },
@@ -48,13 +49,14 @@ export async function ConnectedClusterCard({
   return (
     <ClusterCard
       isLoading={false}
-      status={res?.cluster.attributes.status || "n/a"}
+      status={res?.cluster.attributes.status ?? "n/a"}
       messages={messages ?? []}
       name={res?.cluster.attributes.name || "n/a"}
       consumerGroups={groupCount}
       brokersOnline={brokersOnline}
       brokersTotal={brokersTotal}
-      kafkaVersion={res?.cluster.attributes.kafkaVersion || "n/a"}
+      kafkaVersion={res?.cluster.attributes.kafkaVersion ?? "n/a"}
+      kafkaId={res.cluster.id}
     />
   );
 }

ui/app/[locale]/(authorized)/kafka/[kafkaId]/overview/page.tsx

@@ -30,6 +30,7 @@ export default function OverviewPage({ params }: { params: KafkaParams }) {
   const viewedTopics = getViewedTopics().then((topics) =>
     topics.filter((t) => t.kafkaId === params.kafkaId),
   );
+
   return (
     <PageLayout
       clusterOverview={

ui/app/[locale]/(authorized)/layout.tsx

@@ -2,8 +2,6 @@ import { getAuthOptions } from "@/app/api/auth/[...nextauth]/route";
 
 import { getServerSession } from "next-auth";
 import { ReactNode } from "react";
-import { AppLayout } from "../../../components/AppLayout";
-import { AppLayoutProvider } from "../../../components/AppLayoutProvider";
 import { AppSessionProvider } from "./AppSessionProvider";
 import { SessionRefresher } from "./SessionRefresher";
 

ui/components/AppLayout.tsx

@@ -6,13 +6,22 @@ import { AppSidebar } from "./AppSidebar";
 import { ClusterDrawer } from "./ClusterDrawer";
 
 import { ClusterDrawerProvider } from "./ClusterDrawerProvider";
+import { ReconciliationProvider } from "./ReconciliationProvider";
+import { ReconciliationPausedBanner } from "./ReconciliationPausedBanner";
 
 export function AppLayout({
   username,
   sidebar,
   children,
-}: PropsWithChildren<{ username?: string; sidebar?: ReactNode }>) {
+  kafkaId,
+}: PropsWithChildren<{
+  username?: string;
+  sidebar?: ReactNode;
+  kafkaId?: string;
+}>) {
   const t = useTranslations();
+
+  const isValidKafkaId = !!kafkaId;
   return (
     <Page
       header={<AppMasthead username={username} showSidebarToggle={!!sidebar} />}
@@ -28,7 +37,12 @@ export function AppLayout({
     >
       {/*<HelpContainer>*/}
       <ClusterDrawerProvider>
-        <ClusterDrawer>{children}</ClusterDrawer>
+      <ReconciliationProvider kafkaId={kafkaId ?? ""}>
+          {isValidKafkaId && (
+            <ReconciliationPausedBanner kafkaId={kafkaId} />
+          )}
+          <ClusterDrawer>{children}</ClusterDrawer>
+        </ReconciliationProvider>
       </ClusterDrawerProvider>
       {/*</HelpContainer>*/}
     </Page>

ui/components/ClusterOverview/ClusterCard.stories.tsx

@@ -1,9 +1,22 @@
 import type { Meta, StoryObj } from "@storybook/react";
 
 import { ClusterCard } from "./ClusterCard";
+import { ReconciliationContext } from "../ReconciliationContext";
 
 const meta: Meta<typeof ClusterCard> = {
   component: ClusterCard,
+  decorators: [
+    (Story) => (
+      <ReconciliationContext.Provider
+        value={{
+          isReconciliationPaused: true,
+          setReconciliationPaused: () => { }
+        }}
+      >
+        <Story />
+      </ReconciliationContext.Provider>
+    )
+  ]
 };
 
 export default meta;