feat: Large File Support — proxy, operator, SDKs, and demos

[kamir]

Summary

Complete LFS (Large File Support) subsystem for streaming large binary payloads through KafScale via S3-backed object storage with Kafka metadata references.

Core LFS Proxy (119 files)

• LFS proxy server (cmd/lfs-proxy/) — HTTP upload/download, S3 backend, Kafka record publishing, Prometheus metrics, checksum verification, TLS/SASL support, OpenAPI spec
• LFS package (pkg/lfs/) — shared types, producer, consumer, resolver, envelope, and S3 client logic
• Operator integration — reconcile LFS proxy Deployment, Service, and ConfigMap as part of KafscaleCluster CR
• Console LFS admin (internal/console/) — object/topic/orphan browsing handlers
• IDoc exploder (cmd/idoc-explode/) — SAP IDoc payload processing with field aggregation into routed parent segments
• Iceberg processor — LFS-aware decoding and config extensions
• Helm charts — LFS proxy deployment, metrics, monitoring, and ServiceMonitor templates
• E2E tests — LFS proxy broker, HTTP, etcd, SDK, and iceberg processor tests
• CI pipeline — Docker build for LFS proxy image, stage-release workflow
• Documentation — architecture, data-flow, security hardening, Helm deployment, SDK roadmap

Client SDKs & Demos (97 files)

• Java SDK — retry/backoff, configurable timeouts, Maven build
• Python SDK — retry/backoff, envelope codec, pip-installable
• JavaScript SDK (Node.js) — streaming upload support
• Browser SDK (JS) — SPA-based LFS file management
• Demos — E60-E62 (industry designs), E70 (Java), E71 (Python), E72 (browser SPA)
• Docker Compose stack for local LFS development

Fixes included

• Fix missing time import in lfs_handlers.go
• Fix Makefile script name mismatches (lfs-demo-medical.sh → medical-lfs-demo.sh)
• Add missing LFS demo scripts not included in initial branch split
• Fix wrong LFS envelope field names in demo scripts (lfs_version → kfs_lfs)
• Enrich IDoc exploder with Fields map for parent segment field aggregation
• Enrich IDoc demo with full LFS pipeline (S3 upload → envelope → resolve → explode)
• Add Apache 2.0 license headers to 12 files
• Fix S3 blob verification in medical/video/industrial demos (*.meta → obj-*)

Test plan

• go vet ./... passes
• go test -race ./... passes
• go build ./cmd/lfs-proxy passes
• License header check passes (hack/check_license_headers.py)
• Helm lint passes (helm lint deploy/helm/kafscale)
• Coverage gate passes (80% floor)
• E2E tests with MinIO + broker (requires cluster infrastructure)
• Helm chart deployment validation on kind cluster

🤖 Generated with Claude Code

[klaudworks]

The retry logic reuses the same io.Reader across attempts. Once the first attempt reads from it, the data is gone. Retries upload an empty or truncated payload to S3, a pointer record gets written to Kafka, and no error is returned. This silently corrupts the data.

[kamir]

Bug: Produce and ProducePartitioned passed the same io.Reader to
every retry attempt. After the first attempt consumed the reader,
retries uploaded empty/truncated data to S3, and the pointer
record was written to Kafka with no error — silent data
corruption.

Fix: Added replayable() and rewind() helpers:

• replayable(r) — if the reader already implements io.Seeker
  (e.g. *os.File, *bytes.Reader), returns it as-is. Otherwise
  buffers the contents into a *bytes.Reader so it can be replayed.
• rewind(r) — seeks back to offset 0 before each retry attempt.

Both Produce (line 137) and ProducePartitioned (line 199) now
call replayable before the retry loop and rewind at the top of
each retry iteration. This ensures every attempt uploads the full
payload.

[klaudworks]

one foundational problem I see: the lfs-proxy reimplements the whole logic in the current proxy:

Reimplemented from the existing proxy:

• Kafka protocol loop (read request, dispatch by API type, write response)
• Locally handled APIs (ApiVersions, Metadata with host rewriting, FindCoordinator)
• Backend management (discovery from etcd, cached backend list, round-robin selection, retry/backoff, health tracking)
• Backend security (TLS wrapping, SASL authentication)
• Error handling ("not ready" responses, reconnect on failure)

Dropped from the existing proxy:

• sending each partition's data to the broker that owns it
• Connection pooling (reusing backend connections across requests)
• Kafka protocol edge cases (acks=0, retry on "not leader", fan-out to multiple brokers

The consequence is that we have to maintain 2 parallel proxies doing largely the same thing. The lfs-proxy also reintroduces the following bugs:

#124, #122, #121, #120

[klaudworks]

one foundational problem I see: the lfs-proxy reimplements the whole logic in the current proxy:

Reimplemented from the existing proxy:

• Kafka protocol loop (read request, dispatch by API type, write response)
• Locally handled APIs (ApiVersions, Metadata with host rewriting, FindCoordinator)
• Backend management (discovery from etcd, cached backend list, round-robin selection, retry/backoff, health tracking)
• Backend security (TLS wrapping, SASL authentication)
• Error handling ("not ready" responses, reconnect on failure)

Dropped from the existing proxy:

• sending each partition's data to the broker that owns it
• Connection pooling (reusing backend connections across requests)
• Kafka protocol edge cases (acks=0, retry on "not leader", fan-out to multiple brokers

The consequence is that we have to maintain 2 parallel proxies doing largely the same thing. The lfs-proxy also reintroduces the following bugs:

#124, #122, #121, #120

I think the LFS proxy could be a feature inside the existing proxy but I'll let you and @novatechflow decide that.

[klaudworks]

I can also recommend a review agent like this one. I am using opencode but its probably easily adaptable to be used in Claude Code. Running that iteratively until the agent stops complaining helped me to find quite a lot of issues in my prior PRs.

---
description: Performs thorough code review of current changes, looking for regressions, subtle bugs, and design issues like a staff engineer.
mode: primary
temperature: 0.1
tools:
  write: false
  edit: false
  bash: true
permission:
  bash:
    "*": deny
    "grep *"
    "go test*": allow
    "go build*": allow
    "go vet*": allow
    "go fmt*": allow
    "git *": allow
    "git push *": deny
    "git reset *": deny
---

You are a staff-level engineer performing a thorough review of the current changes in this repository.

Your job is to identify problems and simplification opportunities, not to make changes. Read the diff carefully, then explore the surrounding code to understand the full context before forming opinions. Before diving into details, step back and question whether the approach itself makes sense — does it actually achieve its intended goal, and is there a fundamentally better way to solve the same problem? Assume tests and the build already pass. Flag complexity issues in surrounding code when the change interacts with it, but do not suggest improvements completely unrelated to the changes under review.

Focus on:

- **Regressions**: Does this change break existing behavior? Look at callers, tests, and interfaces that depend on modified code.
- **Concurrency issues**: Race conditions, missing locks, unsafe shared state, goroutine leaks.
- **Error handling**: Swallowed errors, missing error propagation, inconsistent error behavior compared to neighboring code.
- **Edge cases**: Nil pointers, empty slices, integer overflow, off-by-one errors, zero-value traps.
- **Contract violations**: Does the change respect the implicit contracts of the interfaces and functions it touches? Are invariants preserved?
- **Resource leaks**: Unclosed connections, files, or channels. Missing deferred cleanup.
- **Behavioral inconsistencies**: Does the new code behave differently from similar patterns already in the codebase?
- **Architecture and complexity**: Does the change introduce or reveal tight coupling, layering violations, misplaced responsibilities, unnecessary indirection, redundant abstractions, or duplicated logic? Would the change be significantly simpler under a different structural arrangement? Could touched code paths be expressed more directly?
- **Test value**: Are added tests low-value (testing trivial behavior, duplicating existing coverage, or tightly coupled to implementation details)? Are there overlapping tests that could be consolidated? Are high-value tests missing — particularly for edge cases, error paths, concurrency, and integration boundaries that the change affects?

Present your findings in two sections:

## Issues

Numbered list sorted by impact. For each: location (file:line), what is wrong and how it manifests, severity (critical/high/medium/low), and a concrete recommendation.

## Simplification Opportunities

Numbered list sorted by impact. For each: what is unnecessarily complex, where, what a simpler version looks like, and what improves as a result.

If either section has no items, say so explicitly. Do not invent problems or fabricate opportunities.

[novatechflow]

@klaudworks we thought about that - but the LFS Proxy needs an SDK as a producer - the normal proxy does not know what file size comes. Means we'd lose default Kafka compatibility. So @kamir's idea was to have a different proxy as an add-on since not all implementations need large file support. Whats your take? We have a discussion open about this feature => #87

[klaudworks]

@klaudworks we thought about that - but the LFS Proxy needs an SDK as a producer - the normal proxy does not know what file size comes. Means we'd lose default Kafka compatibility. So @kamir's idea was to have a different proxy as an add-on since not all implementations need large file support. Whats your take? We have a discussion open about this feature => #87

My main concern is maintaining 2 different versions of kafka plumbing and reintroducing the already fixed bugs, race conditions etc. Also going forward each bug / feature needs to be implemented in both proxies.

According to my current understanding, the normal proxy could just have a feature flag for the LFS. If that is enabled the proxy rewrites messages with the LFS_BLOB header as it currently does (currently it also just rewrites messages with this header).

The current proxy could then be horizontally scaled if needed or if you want strict separation you can have a separate proxy instance with lfs support enabled where kafka clients send their large messages.

I am assuming that @kamir would like to merge this soonish. So I'd recommend to

1. either copy this thread into claude code for a rewrite into the current proxy, then run a few review iterations
2. or to leave it as a separate component for now. The PR is mainly additive so it doesn't mess with other use cases. IMO this is no long term solution because the current lfs-proxy status will be buggy. However, it may work reliably for the happy case of lfs proxy + single broker + single producer + single consumer. You can also integrate it into the proxy later.

[novatechflow]

We should not merge when we might find a better solution and avoid technical debt. That's why I pointed to the discussion; that's the board to discuss this from an engineering level. The feature flag is quite charming - can be set dynamically with the proxy YAML.

[kamir]

Update: LFS Proxy Merged into Unified Proxy

This commit merges the standalone LFS proxy (cmd/lfs-proxy/) into the existing Kafka proxy (cmd/proxy/) as a feature-flagged module behind KAFSCALE_PROXY_LFS_ENABLED (default: OFF).

Architecture

The integration inserts LFS record rewriting between parse and route in handleProduceRouting():

Parse ProduceRequest → [LFS rewrite records in-place] → groupByBroker → fanOut → merge

When payload is set to nil after rewrite, fanOutProduce() automatically re-encodes via protocol.EncodeProduceRequest() — the same mechanism already used for NOT_LEADER retries. No changes to the existing fan-out/routing machinery.

Changes to cmd/proxy/main.go (3 surgical edits)

1. Added lfs *lfsModule field to proxy struct
2. Wired initLFSModule() into main() behind feature flag + HTTP/metrics server startup
3. Inserted 12-line LFS check in handleProduceRouting() (between parse and acks=0 check)
4. Added p.lfs.Shutdown() to cleanup

New Files (16 files, ~4,900 lines)

File	Purpose
lfs.go	lfsModule struct, initLFSModule(), rewriteProduceRequest() adapter, health check, backend connectivity
lfs_rewrite.go	rewriteProduceRecords() — batch decode/encode, LFS_BLOB header detection, S3 upload, envelope replacement
lfs_record.go	Record encoding, varint helpers, lfsBuildRecordBatch()
lfs_s3.go	s3Uploader — PutObject, multipart upload, presign, stream upload
lfs_http.go	HTTP API (/lfs/produce, /lfs/download, /lfs/uploads/*) with chunked upload sessions
lfs_metrics.go	Prometheus metrics (upload duration, bytes, errors, orphans, runtime stats)
lfs_tracker.go + lfs_tracker_types.go	Async Kafka-based operations event log with circuit breaker
lfs_sasl_encode.go	SASL handshake encoding + produce request encoding for HTTP API path
lfs_backend_auth.go	Backend TLS wrapping + SASL PLAIN auth for HTTP API connections
lfs_backend_tls.go	Backend TLS config builder
lfs_http_tls.go	HTTP server TLS config builder
lfs_swagger.go	Swagger UI + OpenAPI spec handler
lfs_uuid.go	UUID generation
openapi.yaml	Embedded OpenAPI spec
lfs_test.go	9 tests with in-memory fakeS3 backend

Review Issues Addressed

All 7 issues identified during code review have been fixed:

1. lfsDropHeader input slice mutation — Changed from headers[:0] (which shares the underlying array and could corrupt other references) to make([]kmsg.Header, 0, len(headers)) which allocates a fresh slice.

2. connectBackend reads env vars on every call — Moved KAFSCALE_LFS_PROXY_BACKEND_RETRIES and KAFSCALE_LFS_PROXY_BACKEND_BACKOFF_MS parsing to initLFSModule(), stored as backendRetries and backendBackoff struct fields on lfsModule. No more os.Getenv + strconv.Atoi per connection attempt.

3. forwardToBackend unused addr parameter — Removed the unused addr string parameter from the method signature. Updated both callers in lfs_http.go (lines 349, 859) to use _ for the addr returned by connectBackend.

4. Orphan tracking gap in proxy path — rewriteProduceRequest now returns (bool, []orphanInfo, error) instead of (bool, error). The caller in handleProduceRouting() holds the orphan list and only calls p.lfs.trackOrphans(lfsOrphans) if the downstream forwardProduce() returns an error. Previously, orphans were collected but silently discarded.

5. No deadline on forwardToBackend — Added conn.SetDeadline() using the context deadline (if available) or m.dialTimeout as fallback. Deadline is cleared after the read/write completes via defer.

6. Test csErr unused variable — Replaced bytes.Contains([]byte(err.Error()), []byte("checksum")) with proper errors.As(err, &csErr) type assertion plus field validation (csErr.Expected). Removed unused bytes import.

7. lfsGetClientIP bool named err — Renamed the bool return from strings.Cut from err to found for clarity.

What Stays Unchanged

• cmd/lfs-proxy/ — untouched except for deprecation warning log
• pkg/lfs/ — untouched
• All demo scripts and Makefile targets — untouched (they validate the refactoring)
• Existing proxy tests — all 20 pass unchanged
• All pkg/ tests pass

Test Results

cmd/proxy/     — 29 tests PASS (20 existing + 9 LFS)
cmd/lfs-proxy/ — PASS (untouched)
pkg/lfs/       — PASS (untouched)
go vet ./...   — clean
go test -race  — clean

[novatechflow]

@klaudworks @kamir - can you please resolve the comments?

[novatechflow]

Please cleanup

[kamir]

Superseded by #134 — a surgically clean branch (feat/lfs-core-clean) containing only the 212 core LFS feature files.

What changed:

• Removed all demo scripts, deployment guides, staging infra, spring-boot examples, node_modules, and personal config that were accidentally included here
• Core LFS functionality is identical

The original lfs-core branch is preserved for cherry-picking demo/deployment material into a future separate PR.