linter fixes;

Author: Jan Kamieth

backend/.gitignore

@@ -1,4 +1,6 @@
-!config.dist.yml
 config.*.yml
+!config.dist.yml
 public/*
+!public/index.html
+!public/fav.png
 /flink-admin

backend/build/Dockerfile

@@ -0,0 +1,46 @@
+# Stage 1: Build the frontend
+FROM oven/bun:1 AS frontend-builder
+WORKDIR /app/frontend
+
+# Copy package files first to leverage Docker cache
+COPY frontend/package.json frontend/bun.lock ./
+RUN bun install --frozen-lockfile
+
+# Copy the rest of the frontend source code
+COPY frontend/ ./
+# Build the frontend (outputs to /app/frontend/dist)
+RUN bun run build
+
+# Stage 2: Build the backend with embedded frontend
+FROM golang:1.25 AS backend-builder
+WORKDIR /app
+
+# Copy backend module files first
+COPY backend/go.mod backend/go.sum ./
+RUN go mod download
+
+# Copy backend source code
+COPY backend/ ./
+
+# Copy the built frontend artifacts into the backend source tree
+# This is crucial for the //go:embed directive to pick them up
+COPY --from=frontend-builder /app/frontend/dist ./public
+
+# Build the Go binary
+# CGO_ENABLED=0 creates a statically linked binary
+RUN CGO_ENABLED=0 GOOS=linux go build -o flink-admin .
+
+# Stage 3: Create the final runtime image
+# distroless/static is ideal for static Go binaries (small, secure)
+FROM gcr.io/distroless/static-debian12 AS runtime
+
+WORKDIR /app
+
+# Copy the binary from the builder stage
+COPY --from=backend-builder /app/flink-admin .
+
+# Expose the port configured in config.dist.yml
+EXPOSE 8081
+
+# Run the binary
+ENTRYPOINT ["./flink-admin"]

backend/build/Dockerfile.release

@@ -0,0 +1,27 @@
+# Dockerfile.release - Packaging-only Dockerfile for CI multi-arch builds
+# This Dockerfile contains NO RUN commands, allowing Buildx to assemble
+# images for any target platform without QEMU emulation.
+#
+# Prerequisites (provided by CI workflow):
+#   - Pre-built flink-admin binary at: dist/<arch>/flink-admin
+#   - Migrations in repo at: backend/build/migrations/
+#
+# Usage:
+#   docker buildx build --platform linux/amd64 \
+#     --build-arg TARGETARCH=amd64 \
+#     -f build/Dockerfile.release .
+
+FROM gcr.io/distroless/static-debian12
+
+ARG TARGETARCH
+
+WORKDIR /app
+
+# Copy the pre-built binary for the target architecture
+COPY dist/${TARGETARCH}/flink-admin ./flink-admin
+
+# Expose the application port
+EXPOSE 8081
+
+# Run the binary
+ENTRYPOINT ["./flink-admin"]

backend/config.dist.yml

@@ -0,0 +1,43 @@
+env: "sandbox"
+
+app_project: "justtrack"
+app_family: "gosoline"
+app_group: "flink"
+app_name: "admin"
+
+aws:
+  account_id: "047452524847"
+  organizational_unit: "sdlc"
+
+cloud:
+  aws:
+    defaults:
+      endpoint: ""
+      profile: "{aws.organizational_unit}-marketing"
+    s3:
+      clients:
+        default:
+          region: eu-central-1
+
+flink:
+  namespaces:
+    - annotators
+    - denormalizers
+    - lake-ingesters
+    - parquetizers
+    - session-tracker
+
+httpserver:
+  default:
+    port: 8082
+    timeout:
+      read: 60s
+      write: 60s
+      idle: 60s
+    compression:
+      exclude:
+        path:
+          - /api/deployments/watch
+
+kube:
+  context: "arn:aws:eks:eu-central-1:{aws.account_id}:cluster/{aws.organizational_unit}-marketing"

backend/internal/checkpoint/checkpoint_properties.go

@@ -58,5 +58,6 @@ func binaryBigEndianUint16(data []byte) uint16 {
 	if len(data) < 2 {
 		return 0
 	}
+
 	return uint16(data[0])<<8 | uint16(data[1])
 }

backend/internal/checkpoint/keyed_state.go

@@ -181,65 +181,28 @@ func readHandleAndLocalPathList(br *binaryReader) ([]HandleAndLocalPath, error)
 // Type 14 (V2) includes a separate checkpointId field.
 func readChangelogStateHandle(br *binaryReader, kind KeyedStateHandleType, parseFull bool) (KeyedStateHandle, error) {
 	_ = parseFull
-	isV2 := kind == KeyedStateHandleChangelogV2
-
-	startKeyGroup, err := br.ReadInt32()
-	if err != nil {
-		return nil, fmt.Errorf("read changelog start key group: %w", err)
-	}
-	count, err := br.ReadInt32()
-	if err != nil {
-		return nil, fmt.Errorf("read changelog key group count: %w", err)
-	}
-	checkpointedSize, err := br.ReadInt64()
+	startKeyGroup, numKeyGroups, checkpointedSize, err := readChangelogHeader(br)
 	if err != nil {
-		return nil, fmt.Errorf("read changelog checkpointed size: %w", err)
+		return nil, err
 	}
 
-	materializedCount, err := br.ReadInt32()
+	materialized, err := readChangelogKeyedStateHandles(br, "materialized")
 	if err != nil {
-		return nil, fmt.Errorf("read changelog materialized count: %w", err)
-	}
-	if materializedCount < 0 {
-		return nil, fmt.Errorf("changelog materialized count negative: %d", materializedCount)
-	}
-	materialized := make([]KeyedStateHandle, 0, materializedCount)
-	for i := int32(0); i < materializedCount; i++ {
-		handle, err := readKeyedStateHandle(br, true)
-		if err != nil {
-			return nil, fmt.Errorf("read changelog materialized handle: %w", err)
-		}
-		if handle != nil {
-			materialized = append(materialized, handle)
-		}
+		return nil, err
 	}
 
-	nonMaterializedCount, err := br.ReadInt32()
+	nonMaterialized, err := readChangelogKeyedStateHandles(br, "non materialized")
 	if err != nil {
-		return nil, fmt.Errorf("read changelog non materialized count: %w", err)
-	}
-	if nonMaterializedCount < 0 {
-		return nil, fmt.Errorf("changelog non materialized count negative: %d", nonMaterializedCount)
-	}
-	nonMaterialized := make([]KeyedStateHandle, 0, nonMaterializedCount)
-	for i := int32(0); i < nonMaterializedCount; i++ {
-		handle, err := readKeyedStateHandle(br, true)
-		if err != nil {
-			return nil, fmt.Errorf("read changelog non materialized handle: %w", err)
-		}
-		if handle != nil {
-			nonMaterialized = append(nonMaterialized, handle)
-		}
+		return nil, err
 	}
 
 	materializationID, err := br.ReadInt64()
 	if err != nil {
 		return nil, fmt.Errorf("read changelog materialization id: %w", err)
 	}
 
-	// checkpointId is only present in V2 (type 14); for V1, use materializationID
 	checkpointID := materializationID
-	if isV2 {
+	if kind == KeyedStateHandleChangelogV2 {
 		checkpointID, err = br.ReadInt64()
 		if err != nil {
 			return nil, fmt.Errorf("read changelog checkpoint id: %w", err)
@@ -254,7 +217,7 @@ func readChangelogStateHandle(br *binaryReader, kind KeyedStateHandleType, parse
 	return ChangelogStateHandle{
 		Type:              kind,
 		StartKeyGroup:     startKeyGroup,
-		NumKeyGroups:      count,
+		NumKeyGroups:      numKeyGroups,
 		CheckpointedSize:  checkpointedSize,
 		Materialized:      materialized,
 		NonMaterialized:   nonMaterialized,
@@ -264,6 +227,46 @@ func readChangelogStateHandle(br *binaryReader, kind KeyedStateHandleType, parse
 	}, nil
 }
 
+func readChangelogHeader(br *binaryReader) (startKeyGroup int32, numKeyGroups int32, checkpointedSize int64, err error) {
+	startKeyGroup, err = br.ReadInt32()
+	if err != nil {
+		return 0, 0, 0, fmt.Errorf("read changelog start key group: %w", err)
+	}
+	count, err := br.ReadInt32()
+	if err != nil {
+		return 0, 0, 0, fmt.Errorf("read changelog key group count: %w", err)
+	}
+	checkpointedSize, err = br.ReadInt64()
+	if err != nil {
+		return 0, 0, 0, fmt.Errorf("read changelog checkpointed size: %w", err)
+	}
+
+	return startKeyGroup, count, checkpointedSize, nil
+}
+
+func readChangelogKeyedStateHandles(br *binaryReader, label string) ([]KeyedStateHandle, error) {
+	count, err := br.ReadInt32()
+	if err != nil {
+		return nil, fmt.Errorf("read changelog %s count: %w", label, err)
+	}
+	if count < 0 {
+		return nil, fmt.Errorf("changelog %s count negative: %d", label, count)
+	}
+
+	handles := make([]KeyedStateHandle, 0, count)
+	for i := int32(0); i < count; i++ {
+		handle, err := readKeyedStateHandle(br, true)
+		if err != nil {
+			return nil, fmt.Errorf("read changelog %s handle: %w", label, err)
+		}
+		if handle != nil {
+			handles = append(handles, handle)
+		}
+	}
+
+	return handles, nil
+}
+
 // readChangelogByteIncrementHandle parses in-memory changelog increments.
 func readChangelogByteIncrementHandle(br *binaryReader, kind KeyedStateHandleType, parseFull bool) (KeyedStateHandle, error) {
 	startKeyGroup, err := br.ReadInt32()
@@ -304,6 +307,7 @@ func readChangelogByteIncrementHandle(br *binaryReader, kind KeyedStateHandleTyp
 			if _, err := br.ReadBytes(int(length)); err != nil {
 				return nil, fmt.Errorf("read changelog byte data: %w", err)
 			}
+
 			continue
 		}
 		data, err := br.ReadBytes(int(length))

backend/internal/checkpoint/metadata.go

@@ -104,25 +104,43 @@ func ParseSummary(reader io.Reader, options ParseOptions) (*CheckpointSummary, e
 }
 
 // ParseFile opens the given file path and parses it as _metadata.
-func ParseFile(path string, options ParseOptions) (*CheckpointMetadata, error) {
+func ParseFile(path string, options ParseOptions) (metadata *CheckpointMetadata, err error) {
 	file, err := os.Open(path)
 	if err != nil {
 		return nil, fmt.Errorf("open metadata file: %w", err)
 	}
-	defer file.Close()
+	defer func() {
+		if cerr := file.Close(); cerr != nil && err == nil {
+			err = fmt.Errorf("close metadata file: %w", cerr)
+		}
+	}()
 
-	return Parse(file, options)
+	metadata, err = Parse(file, options)
+	if err != nil {
+		return nil, err
+	}
+
+	return metadata, nil
 }
 
 // ParseFileSummary opens the given file path and parses it as a summary.
-func ParseFileSummary(path string, options ParseOptions) (*CheckpointSummary, error) {
+func ParseFileSummary(path string, options ParseOptions) (summary *CheckpointSummary, err error) {
 	file, err := os.Open(path)
 	if err != nil {
 		return nil, fmt.Errorf("open metadata file: %w", err)
 	}
-	defer file.Close()
+	defer func() {
+		if cerr := file.Close(); cerr != nil && err == nil {
+			err = fmt.Errorf("close metadata file: %w", cerr)
+		}
+	}()
+
+	summary, err = ParseSummary(file, options)
+	if err != nil {
+		return nil, err
+	}
 
-	return ParseSummary(file, options)
+	return summary, nil
 }
 
 // readMasterStates parses master state entries from the stream.
@@ -350,6 +368,7 @@ func readOptionalOperatorStateHandle(br *binaryReader, parseFull bool) (*Operato
 	if err != nil {
 		return nil, err
 	}
+
 	return handle, nil
 }
 
@@ -360,5 +379,6 @@ func buildOperatorID(low int64, high int64) [16]byte {
 		id[i] = byte(high >> uint(56-8*i))
 		id[i+8] = byte(low >> uint(56-8*i))
 	}
+
 	return id
 }