Stream Processing Flink Topology

blockdevops

Apache Flink & RisingWave stateful streaming graph: real-time DAG topology, window throughput, checkpointing duration, RocksDB state inspector, and backpressure heat monitor.

Installation

$pnpm dlx shadcn@latest add https://uipkge.dev/r/react/stream-processing-flink-topology.json

$npx shadcn@latest add https://uipkge.dev/r/react/stream-processing-flink-topology.json

$yarn dlx shadcn@latest add https://uipkge.dev/r/react/stream-processing-flink-topology.json

$bunx shadcn@latest add https://uipkge.dev/r/react/stream-processing-flink-topology.json

Named registry:npx shadcn@latest add @uipkge-react/stream-processing-flink-topologyInstalls to:components/blocks/

Variants

Loading interactive previews…

Props

Name	Type / Values	Default	Required
`operators`	`FlinkOperatorNode[]`	—	optional
`className`	`string`	—	optional

Schema

Type aliases exported from this item's source. Use these to shape the data you pass in.

FlinkSubtask

interface FlinkSubtask {
  id: number
  host: string
  status: 'RUNNING' | 'DEPLOYING' | 'RECONCILING'
  inRate: string
  outRate: string
  cpu: string
  heapMemory: string
  backpressure: number
  watermarkLag: string
  checkpointAckMs: number
}

FlinkOperatorNode

interface FlinkOperatorNode {
  id: string
  name: string
  shortTitle: string
  stageNumber: number
  category: 'source' | 'window' | 'inference' | 'sink'
  operatorClass: string
  parallelism: number
  status: 'RUNNING' | 'CANCELED' | 'FAILING'
  backpressure: number
  inputRecordsRate: string
  inputBytesRate: string
  outputRecordsRate: string
  outputBytesRate: string
  totalRecordsIn: number
  totalRecordsOut: number
  managedStateSize: string
  stateBackend: string
  heapMemory: string
  managedMemory: string
  gcTime: string
  cpuPeak: string
  watermarkLag: string
  currentWatermark: string
  description: string
  config: Record<string, string>
  subtasks: FlinkSubtask[]
}

npm dependencies

lucide-react

Includes

badge button card progress separator table

Files installed (5)

components/blocks/StreamProcessingFlinkTopology.tsx21.7 kB

'use client'

import * as React from 'react'
import {
  AlertTriangle,
  ArrowRight,
  Camera,
  Check,
  CheckCircle2,
  Clock,
  Copy,
  Cpu,
  Database,
  Loader2,
  Radio,
  Square,
  Workflow,
} from 'lucide-react'
import { cn } from '@/lib/utils'
import { Badge } from '@/components/ui/badge'
import { Button } from '@/components/ui/button'
import { Card, CardContent, CardDescription, CardHeader, CardTitle } from '@/components/ui/card'
import { StreamOperatorDetail } from './StreamOperatorDetail'
import { StreamTopologyTelemetry } from './StreamTopologyTelemetry'
import type { FlinkOperatorNode, FlinkSubtask } from './stream-topology-types'
import { defaultFlinkOperators } from './stream-topology-data'

export type { FlinkOperatorNode, FlinkSubtask }

export interface StreamProcessingFlinkTopologyProps {
  operators?: FlinkOperatorNode[]
  className?: string
}

export function StreamProcessingFlinkTopology({
  operators = defaultFlinkOperators,
  className,
}: StreamProcessingFlinkTopologyProps) {
  const [selectedOperatorId, setSelectedOperatorId] = React.useState<string>('op-ml-scoring')
  const [isTriggeringSavepoint, setIsTriggeringSavepoint] = React.useState<boolean>(false)
  const [savepointNotice, setSavepointNotice] = React.useState<string | null>(null)
  const [isCanceling, setIsCanceling] = React.useState<boolean>(false)
  const [cancelConfirmed, setCancelConfirmed] = React.useState<boolean>(false)
  const [copiedKey, setCopiedKey] = React.useState<string | null>(null)

  const selectedOperator = React.useMemo(() => {
    return operators.find((op) => op.id === selectedOperatorId) ?? operators[0]
  }, [operators, selectedOperatorId])

  const handleTriggerSavepoint = () => {
    if (isTriggeringSavepoint) return
    setIsTriggeringSavepoint(true)
    setSavepointNotice(null)

    setTimeout(() => {
      setIsTriggeringSavepoint(false)
      setSavepointNotice('s3://flink-savepoints/prod/savepoint-fraud-90412-20260821-143245/')
      setTimeout(() => {
        setSavepointNotice((curr) => (curr?.includes('savepoint-fraud-90412') ? null : curr))
      }, 6500)
    }, 1800)
  }

  const handleCancelJob = () => {
    if (isCanceling) return
    setIsCanceling(true)
    setTimeout(() => {
      setIsCanceling(false)
      setCancelConfirmed(true)
      setTimeout(() => {
        setCancelConfirmed(false)
      }, 4500)
    }, 2000)
  }

  const copyText = (key: string, value: string) => {
    if (typeof navigator !== 'undefined' && navigator.clipboard) {
      navigator.clipboard.writeText(value)
      setCopiedKey(key)
      setTimeout(() => {
        setCopiedKey((curr) => (curr === key ? null : curr))
      }, 2000)
    }
  }

  return (
    <div data-slot="stream-processing-flink-topology" className={cn('w-full space-y-6', className)}>
      {/* 1. Header Section */}
      <div className="flex flex-col gap-4 lg:flex-row lg:items-center lg:justify-between">
        <div className="space-y-1.5">
          <div className="flex flex-wrap items-center gap-2.5">
            <div
              className="bg-muted text-muted-foreground border-border flex size-9 shrink-0 items-center justify-center rounded-lg border shadow-xs"
              aria-hidden="true"
            >
              <Workflow className="text-primary size-4.5" />
            </div>
            <div className="flex flex-wrap items-center gap-2">
              <h1 className="text-foreground font-mono text-xl font-bold tracking-tight break-all sm:text-2xl">
                realtime-fraud-scoring-pipeline
              </h1>
              <Badge variant="outline" className="font-mono text-xs">
                Flink 1.19 on Kubernetes
              </Badge>
            </div>
            <Badge variant="success" className="gap-1.5 text-xs font-medium">
              <span className="relative flex size-2 shrink-0">
                <span className="bg-success absolute inline-flex h-full w-full rounded-full opacity-75" />
                <span className="bg-success relative inline-flex size-2 rounded-full" />
              </span>
              <span>Job Running · Checkpointing Healthy</span>
            </Badge>
          </div>
          <p className="text-muted-foreground font-mono text-xs">
            Checkpoint Interval: Every 10s · RocksDB StateBackend · SLA Target: &lt; 50ms Lag · Job ID: flink_job_992014
          </p>
        </div>

        {/* Action Buttons */}
        <div className="flex flex-wrap items-center gap-2">
          <Button
            variant="outline"
            size="sm"
            className="gap-1.5 text-xs font-medium"
            disabled={isTriggeringSavepoint}
            onClick={handleTriggerSavepoint}
          >
            {isTriggeringSavepoint ? (
              <Loader2 className="size-3.5 animate-spin" aria-hidden="true" />
            ) : (
              <Camera className="text-info size-3.5" aria-hidden="true" />
            )}
            <span>{isTriggeringSavepoint ? 'Creating Savepoint...' : 'Trigger Savepoint'}</span>
          </Button>

          <Button
            variant="outline"
            size="sm"
            className="text-destructive hover:border-destructive/40 hover:bg-destructive/10 gap-1.5 text-xs font-medium"
            disabled={isCanceling}
            onClick={handleCancelJob}
          >
            {isCanceling ? (
              <Loader2 className="size-3.5 animate-spin" aria-hidden="true" />
            ) : (
              <Square className="size-3.5 fill-current" aria-hidden="true" />
            )}
            <span>{isCanceling ? 'Canceling with Savepoint...' : 'Cancel Job with Savepoint'}</span>
          </Button>
        </div>
      </div>

      {/* Active Savepoint Notice Banner */}
      {savepointNotice && (
        <div
          className="text-foreground border-info/30 bg-info/10 flex items-center justify-between rounded-lg border p-3 text-xs shadow-xs"
          role="status"
        >
          <div className="flex items-center gap-2.5">
            <CheckCircle2 className="text-info size-4 shrink-0" aria-hidden="true" />
            <div>
              <span className="text-info font-semibold">Savepoint completed successfully:</span>
              <span className="ml-1 font-mono text-xs">{savepointNotice}</span>
            </div>
          </div>
          <Button
            variant="ghost"
            size="xs"
            className="h-6 gap-1 px-2 text-xs"
            onClick={() => copyText('savepoint', savepointNotice)}
          >
            {copiedKey === 'savepoint' ? <Check className="text-success size-3" /> : <Copy className="size-3" />}
            <span>{copiedKey === 'savepoint' ? 'Copied' : 'Copy URI'}</span>
          </Button>
        </div>
      )}

      {/* Active Cancellation Notice Banner */}
      {cancelConfirmed && (
        <div
          className="text-foreground border-warning/30 bg-warning/10 flex items-center justify-between rounded-lg border p-3 text-xs shadow-xs"
          role="status"
        >
          <div className="flex items-center gap-2.5">
            <AlertTriangle className="text-warning size-4 shrink-0" aria-hidden="true" />
            <div>
              <span className="text-warning font-semibold">Graceful Job Drain Initiated:</span>
              <span className="ml-1">
                Stopping topology sources and committing final RocksDB savepoint to S3 storage.
              </span>
            </div>
          </div>
          <Badge variant="outline" className="font-mono text-xs">
            Drain Status: 100%
          </Badge>
        </div>
      )}

      {/* 2. 4 Stream Telemetry KPI Cards */}
      <StreamTopologyTelemetry />

      {/* 3. Interactive 4-Operator Streaming Topology DAG */}
      <Card className="shadow-xs">
        <CardHeader className="pb-3">
          <div className="flex flex-col gap-2 sm:flex-row sm:items-center sm:justify-between">
            <div>
              <div className="flex items-center gap-2">
                <CardTitle className="text-base font-semibold">Streaming Execution Topology DAG</CardTitle>
                <Badge variant="outline" className="font-mono text-xs">
                  4 Stateful Operators
                </Badge>
              </div>
              <CardDescription className="text-xs">
                Live dataflow graph with event-time watermarking, state sizes, and backpressure telemetry. Click any
                operator to inspect metrics and subtasks.
              </CardDescription>
            </div>
            <div className="flex flex-wrap items-center gap-2">
              <div className="border-border bg-muted/40 text-muted-foreground flex items-center gap-1.5 rounded-md border px-2.5 py-1 font-mono text-xs">
                <span className="bg-success size-2 rounded-full" />
                <span>Backpressure Heat: 0.0% (Green)</span>
              </div>
              <div className="border-border bg-muted/40 text-muted-foreground flex items-center gap-1.5 rounded-md border px-2.5 py-1 font-mono text-xs">
                <Database className="size-3" />
                <span>Total Managed State: 57.0 MB</span>
              </div>
            </div>
          </div>
        </CardHeader>

        <CardContent className="p-4 pt-1 sm:p-6">
          <div className="border-border bg-muted/15 rounded-xl border p-4 sm:p-5">
            {/* 4 Connected Nodes Grid */}
            <div className="grid grid-cols-1 gap-4 lg:grid-cols-4">
              {/* Node 1: Kafka Ingestion Source */}
              <div
                role="button"
                tabIndex={0}
                className={cn(
                  'group focus-visible:ring-ring relative flex cursor-pointer flex-col justify-between rounded-xl border p-4 transition-colors duration-200 focus-visible:ring-2 focus-visible:outline-none',
                  selectedOperatorId === 'op-kafka-source'
                    ? 'border-primary bg-primary/[0.04] ring-primary/30 shadow-xs ring-2'
                    : 'border-border bg-card hover:border-primary/50 hover:bg-muted/30',
                )}
                onClick={() => setSelectedOperatorId('op-kafka-source')}
                onKeyDown={(e) => {
                  if (e.key === 'Enter' || e.key === ' ') {
                    e.preventDefault()
                    setSelectedOperatorId('op-kafka-source')
                  }
                }}
              >
                <div className="space-y-3">
                  <div className="flex items-start justify-between gap-2">
                    <div className="flex items-center gap-2">
                      <div className="border-info/30 bg-info/10 text-info flex size-7 items-center justify-center rounded-md border">
                        <Radio className="size-3.5" />
                      </div>
                      <Badge variant="outline" className="border-info/30 text-info font-mono text-xs">
                        SOURCE
                      </Badge>
                    </div>
                    <Badge variant="outline" className="font-mono text-xs">
                      p=8
                    </Badge>
                  </div>

                  <div>
                    <div className="text-foreground font-mono text-xs leading-snug font-bold">
                      Kafka Ingestion Source
                    </div>
                    <p className="text-muted-foreground mt-1 font-mono text-xs">42.5k rec/s · Backpressure: 0%</p>
                  </div>

                  <div className="border-border/60 space-y-1.5 border-t pt-2.5 font-mono text-xs">
                    <div className="text-muted-foreground flex items-center justify-between">
                      <span>Throughput:</span>
                      <span className="text-foreground font-semibold tabular-nums">8.4 MB/s</span>
                    </div>
                    <div className="text-muted-foreground flex items-center justify-between">
                      <span>State Size:</span>
                      <span className="text-foreground tabular-nums">0 MB (Stateless)</span>
                    </div>
                  </div>
                </div>

                <div className="border-border/60 mt-3 flex items-center justify-between border-t pt-2 text-xs">
                  <span className="text-muted-foreground font-mono">events.fraud.transactions</span>
                  <ArrowRight className="text-muted-foreground group-hover:text-primary size-3.5 shrink-0 transition-colors" />
                </div>
              </div>

              {/* Node 2: Keyed Sliding Window */}
              <div
                role="button"
                tabIndex={0}
                className={cn(
                  'group focus-visible:ring-ring relative flex cursor-pointer flex-col justify-between rounded-xl border p-4 transition-colors duration-200 focus-visible:ring-2 focus-visible:outline-none',
                  selectedOperatorId === 'op-keyed-window'
                    ? 'border-primary bg-primary/[0.04] ring-primary/30 shadow-xs ring-2'
                    : 'border-border bg-card hover:border-primary/50 hover:bg-muted/30',
                )}
                onClick={() => setSelectedOperatorId('op-keyed-window')}
                onKeyDown={(e) => {
                  if (e.key === 'Enter' || e.key === ' ') {
                    e.preventDefault()
                    setSelectedOperatorId('op-keyed-window')
                  }
                }}
              >
                <div className="space-y-3">
                  <div className="flex items-start justify-between gap-2">
                    <div className="flex items-center gap-2">
                      <div className="border-chart-1/30 bg-chart-1/10 text-chart-1 flex size-7 items-center justify-center rounded-md border">
                        <Clock className="size-3.5" />
                      </div>
                      <Badge variant="outline" className="border-chart-1/30 text-chart-1 font-mono text-xs">
                        WINDOW
                      </Badge>
                    </div>
                    <Badge variant="outline" className="font-mono text-xs">
                      p=16
                    </Badge>
                  </div>

                  <div>
                    <div className="text-foreground font-mono text-xs leading-snug font-bold">Keyed Sliding Window</div>
                    <p className="text-muted-foreground mt-1 font-mono text-xs">
                      5-min window, 10s slide · 12,400 keys
                    </p>
                  </div>

                  <div className="border-border/60 space-y-1.5 border-t pt-2.5 font-mono text-xs">
                    <div className="text-muted-foreground flex items-center justify-between">
                      <span>Output Rate:</span>
                      <span className="text-foreground font-semibold tabular-nums">38.2k rec/s</span>
                    </div>
                    <div className="text-muted-foreground flex items-center justify-between">
                      <span>RocksDB State:</span>
                      <span className="text-foreground text-chart-1 font-semibold tabular-nums">18.4 MB</span>
                    </div>
                  </div>
                </div>

                <div className="border-border/60 mt-3 flex items-center justify-between border-t pt-2 text-xs">
                  <span className="text-muted-foreground font-mono">Cardholder Velocity Agg</span>
                  <ArrowRight className="text-muted-foreground group-hover:text-primary size-3.5 shrink-0 transition-colors" />
                </div>
              </div>

              {/* Node 3: ML Fraud Scoring Pattern Matcher */}
              <div
                role="button"
                tabIndex={0}
                className={cn(
                  'group focus-visible:ring-ring relative flex cursor-pointer flex-col justify-between rounded-xl border p-4 transition-colors duration-200 focus-visible:ring-2 focus-visible:outline-none',
                  selectedOperatorId === 'op-ml-scoring'
                    ? 'border-primary bg-primary/[0.04] ring-primary/30 shadow-xs ring-2'
                    : 'border-border bg-card hover:border-primary/50 hover:bg-muted/30',
                )}
                onClick={() => setSelectedOperatorId('op-ml-scoring')}
                onKeyDown={(e) => {
                  if (e.key === 'Enter' || e.key === ' ') {
                    e.preventDefault()
                    setSelectedOperatorId('op-ml-scoring')
                  }
                }}
              >
                <div className="space-y-3">
                  <div className="flex items-start justify-between gap-2">
                    <div className="flex items-center gap-2">
                      <div className="border-chart-2/30 bg-chart-2/10 text-chart-2 flex size-7 items-center justify-center rounded-md border">
                        <Cpu className="size-3.5" />
                      </div>
                      <Badge variant="outline" className="border-chart-2/30 text-chart-2 font-mono text-xs">
                        CEP & ML
                      </Badge>
                    </div>
                    <Badge variant="outline" className="font-mono text-xs">
                      p=16
                    </Badge>
                  </div>

                  <div>
                    <div className="text-foreground font-mono text-xs leading-snug font-bold">
                      ML Fraud Scoring Matcher
                    </div>
                    <p className="text-muted-foreground mt-1 font-mono text-xs">Inference: 1.2ms · RocksDB: 38.2 MB</p>
                  </div>

                  <div className="border-border/60 space-y-1.5 border-t pt-2.5 font-mono text-xs">
                    <div className="text-muted-foreground flex items-center justify-between">
                      <span>ONNX Inference:</span>
                      <span className="text-success text-success font-semibold tabular-nums">p99 &lt; 1.2ms</span>
                    </div>
                    <div className="text-muted-foreground flex items-center justify-between">
                      <span>Managed State:</span>
                      <span className="text-foreground text-chart-2 font-semibold tabular-nums">38.2 MB</span>
                    </div>
                  </div>
                </div>

                <div className="border-border/60 mt-3 flex items-center justify-between border-t pt-2 text-xs">
                  <span className="text-muted-foreground font-mono">xgboost_v4.onnx</span>
                  <ArrowRight className="text-muted-foreground group-hover:text-primary size-3.5 shrink-0 transition-colors" />
                </div>
              </div>

              {/* Node 4: Sink: Alert Dispatcher & Redis Cache */}
              <div
                role="button"
                tabIndex={0}
                className={cn(
                  'group focus-visible:ring-ring relative flex cursor-pointer flex-col justify-between rounded-xl border p-4 transition-colors duration-200 focus-visible:ring-2 focus-visible:outline-none',
                  selectedOperatorId === 'op-sink-dispatcher'
                    ? 'border-primary bg-primary/[0.04] ring-primary/30 shadow-xs ring-2'
                    : 'border-border bg-card hover:border-primary/50 hover:bg-muted/30',
                )}
                onClick={() => setSelectedOperatorId('op-sink-dispatcher')}
                onKeyDown={(e) => {
                  if (e.key === 'Enter' || e.key === ' ') {
                    e.preventDefault()
                    setSelectedOperatorId('op-sink-dispatcher')
                  }
                }}
              >
                <div className="space-y-3">
                  <div className="flex items-start justify-between gap-2">
                    <div className="flex items-center gap-2">
                      <div className="border-success/30 bg-success/10 text-success flex size-7 items-center justify-center rounded-md border">
                        <Database className="size-3.5" />
                      </div>
                      <Badge variant="outline" className="border-success/30 text-success font-mono text-xs">
                        SINK
                      </Badge>
                    </div>
                    <Badge variant="outline" className="font-mono text-xs">
                      p=8
                    </Badge>
                  </div>

                  <div>
                    <div className="text-foreground font-mono text-xs leading-snug font-bold">
                      Sink: Alert Dispatcher
                    </div>
                    <p className="text-muted-foreground mt-1 font-mono text-xs">Redis Cache & Webhooks · 42.5k rec/s</p>
                  </div>

                  <div className="border-border/60 space-y-1.5 border-t pt-2.5 font-mono text-xs">
                    <div className="text-muted-foreground flex items-center justify-between">
                      <span>Semantics:</span>
                      <span className="text-foreground font-semibold">2PC Exactly-Once</span>
                    </div>
                    <div className="text-muted-foreground flex items-center justify-between">
                      <span>Fan-Out:</span>
                      <span className="text-foreground font-semibold tabular-nums">42.5k rec/s</span>
                    </div>
                  </div>
                </div>

                <div className="border-border/60 mt-3 flex items-center justify-between border-t pt-2 text-xs">
                  <span className="text-muted-foreground font-mono">redis:6379 + alerts.queue</span>
                  <CheckCircle2 className="text-success size-3.5 shrink-0" />
                </div>
              </div>
            </div>
          </div>
        </CardContent>
      </Card>

      {/* 4. Operator Detailed Performance & State Inspector Panel */}
      <StreamOperatorDetail operator={selectedOperator} copiedKey={copiedKey} onCopy={copyText} />
    </div>
  )
}

components/blocks/StreamOperatorDetail.tsx18.1 kB

components/blocks/StreamTopologyTelemetry.tsx7 kB

components/blocks/stream-topology-types.ts0.9 kB

components/blocks/stream-topology-data.ts19.2 kB

Raw manifest:https://uipkge.dev/r/react/stream-processing-flink-topology.json

Change language