feat(iceberg): single JVM per process instead of per stream-chunk by vikaxsh · Pull Request #962 · datazip-inc/olake

vikaxsh · 2026-05-26T10:49:08Z

Description

This PR refactors the Iceberg writer architecture to use a single shared JVM per OLake CLI invocation instead of creating one JVM per writer thread/chunk.

Previously, every Iceberg writer initialization spawned a dedicated JVM-backed Iceberg client and gRPC server, resulting in significant memory overhead under concurrent backfill and CDC workloads. Large syncs could create many JVM processes, leading to excessive memory consumption and potential OOM issues.

Changes

Introduced a shared JVM architecture where all streams and chunks communicate with a single JVM instance.
Added ThreadSession-based isolation to maintain per-stream/per-chunk state within the shared JVM.
Moved stream-specific configuration from JVM startup arguments to gRPC request metadata:
- Namespace
- Upsert mode
- Identifier field creation
- Iceberg partition transforms
Added StreamMetaCtx on the Go side to propagate stream-specific metadata with every request.
Added ThreadSession management in Java to maintain isolated:
- Iceberg table handles
- Table operators
- Writers
- Commit state
Added CLOSE_SESSION RPC operation for explicit session cleanup and resource release.
Retained catalog initialization and other truly global resources at the JVM level.

Benefits

JVM heap is allocated only once per OLake run.
Significantly reduces memory consumption during concurrent syncs.
Eliminates excessive JVM process creation.
Preserves stream-level isolation through session-based state management.
Improves scalability for parallel backfill and CDC workloads.

Fixes # (issue)

Type of change

Bug fix (non-breaking change which fixes an issue)
New feature (non-breaking change which adds functionality)
Breaking change (fix or feature that would cause existing functionality to not work as expected)
This change requires a documentation update

How Has This Been Tested?

Validated concurrent backfill execution using multiple chunks sharing a single JVM instance.
Validated CDC streams operating concurrently through the shared JVM.
Verified session creation and cleanup via CLOSE_SESSION.
Verified successful table creation, schema evolution, writes, and commits across multiple sessions.
Verified memory utilization remains stable compared to the previous multi-JVM architecture.

Screenshots or Recordings

N/A

Documentation

Documentation Link: [link to README, olake.io/docs, or olake-docs]
N/A (bug fix, refactor, or test changes only)

Related PR's (If Any):

N/A

… into feat/unified-jvm

…se using a cancelled flag

…eat/unified-jvm

hash-data

can we think of better RPC request structure? as well as think of more scenerios where things can break?

hash-data · 2026-06-08T05:57:34Z

+// Shutdown invokes destination-level cleanup if the registered writer
+// implements Shutdownable. No-op for destinations without long-lived
+// resources (parquet).
+func Shutdown(ctx context.Context, config *types.WriterConfig) {


why shutdown? it can be cleanup and it is part of writer pool

hash-data · 2026-06-08T05:58:43Z

+// The catalog/storage portion of `config` is what drives the JVM CLI; later
+// callers that pass a different config still receive the already-running JVM.
+// This is intentional: in a single OLake sync the destination config is fixed.
+func acquireServer(config *Config) (*serverInstance, error) {


think of function name right now it is not readable?
Example:
acquireServer is not what this function does, it create thread in java side so initJavaThread could be a good name

hash-data · 2026-06-08T06:00:12Z

+	sharedServerMu.Lock()
+	defer sharedServerMu.Unlock()



Inside Backfill, for a single stream, chunkProcessor runs concurrently for every chunk of the same stream
so not 2 chunk try to create jvm at same time

hash-data · 2026-06-08T06:01:37Z

 	for attempt := 0; attempt < maxAttempts; attempt++ {
-		// get available port
-		port, err = FindAvailablePort(threadID, nextStartPort)
+		port, err := FindAvailablePort(serverID, nextStartPort)


why we need port here?

if a previous failed sync left behind an orphaned JVM that was still holding the port, this can now be handled easily. kept the existing behavior for now to avoid making significant changes

hash-data · 2026-06-08T09:07:12Z

+		// TODO: research the following flags in arrow writer and legacy writer
+		// need to do some research on the following flags
+		var serverCmd *exec.Cmd
+		if os.Getenv("OLAKE_DEBUG_MODE") != "" {


how this will work if server already started?

in each sync ,server starts once, if provided before sync start, it will work

hash-data · 2026-06-10T05:35:44Z

+  }
+
+  public static Table createIcebergTable(Catalog icebergCatalog, TableIdentifier tableIdentifier, Schema schema) {
+    ensureNamespace(icebergCatalog, tableIdentifier);


do we have exception handling for table already exist?

hash-data · 2026-06-10T05:38:40Z

+                                  BooleanSupplier cancelled) {
+    // Session already torn down before we started — don't create a writer that
+    // nothing will commit/close.
+    if (cancelled.getAsBoolean()) {


will think a better way of it, commenting for reminder

hash-data · 2026-06-10T05:40:01Z

+                    // Only raise the flag — never touch the writer here. This runs on
+                    // a different gRPC thread than the in-flight write, and closing it
+                    // concurrently with writer.write() is what corrupts Parquet. The
+                    // write loop sees `cancelled` and closes its own writer; we don't wait.


memory things need to be removed as there will be a lot of memory hogged by the writer even after close

hash-data · 2026-06-10T05:40:39Z


-    public Table loadIcebergTable(TableIdentifier tableId, Schema schema) {
+    private Table loadOrCreateTable(TableIdentifier tableId, Schema schema, List<Map<String, String>> partitionTransforms) {
        return IcebergUtil.loadIcebergTable(icebergCatalog, tableId).orElseGet(() -> {


we are using iceberg utill directly need to check consequences of it

hash-data · 2026-06-10T05:41:15Z

 import io.grpc.stub.StreamObserver;
 import jakarta.enterprise.context.Dependent;

+/**


not reviewing it as of now, it is going to be similar as olakeRowingestor. A review at end is fine of it

…art per thread session

…eat/unified-jvm

hash-data · 2026-06-12T12:11:21Z

+// receive the already-running JVM. This is intentional: in a single OLake sync
+// the destination config is fixed.
+func initializeServer(config *Config) (*serverInstance, error) {
+	startOnce.Do(func() {


why we need to use DO?

hash-data · 2026-06-17T04:18:40Z

+// Initialize starts destination-level process resources once, up front, if the
+// registered writer implements Initializable. No-op for destinations without
+// long-lived resources (parquet). Pair it with a deferred Shutdown.
+func Initialize(ctx context.Context, config *types.WriterConfig) error {


can we do this in writer pool where we dont need to unmarshal config again, these we can do in this way
-> writer pool start the server
-> the object of server will be global in iceberg and can be used by all instances that get created

hash-data · 2026-06-17T04:19:19Z

+// implements Shutdownable. No-op for destinations without long-lived
+// resources (parquet).
+func Shutdown(ctx context.Context, config *types.WriterConfig) {
+	if config == nil {


shutdown would be writer pool cleanup function where it cleans up the instance of server

hash-data · 2026-06-17T04:20:20Z

+	if err := s.Shutdown(ctx); err != nil {
+		logger.Warnf("destination.Shutdown: %s", err)
+	}
+}


not related to above change:

But can we somehow unmarhsal config only once, right now we are unmarshaling it every time while creating new writer

… into feat/unified-jvm

vikaxsh added 2 commits May 26, 2026 16:18

feat(iceberg): single JVM per process instead of per stream-chunk

77fc64e

Merge branch 'staging' into feat/unified-jvm

f5dce7c

vikaxsh marked this pull request as ready for review May 26, 2026 10:49

vikaxsh added 2 commits May 26, 2026 16:20

chore: dummy commit

96967db

Merge branch 'feat/unified-jvm' of https://github.com/datazip-inc/olake…

2c856ca

… into feat/unified-jvm

vikaxsh marked this pull request as draft May 26, 2026 10:52

vikaxsh added 2 commits May 26, 2026 16:25

chore: undo dummy commit

582f1ff

feat: prevent concurrent writes to the same writer during session clo…

e7b0387

…se using a cancelled flag

vikaxsh requested a deployment to integration_tests June 2, 2026 05:49 — with GitHub Actions Waiting

Merge branch 'staging' of https://github.com/datazip-inc/olake into f…

9fc2f60

…eat/unified-jvm

vikaxsh marked this pull request as ready for review June 2, 2026 05:50

vikaxsh requested a deployment to integration_tests June 2, 2026 05:50 — with GitHub Actions Waiting

fix: lint issue

3c19007

vikaxsh temporarily deployed to integration_tests June 2, 2026 05:57 — with GitHub Actions Inactive

fix: keep thread id consistant in check conenction in dest

d6c5fcc

vikaxsh requested a deployment to integration_tests June 2, 2026 08:01 — with GitHub Actions Waiting

fix: 2pc state refresh issue

98c8259

vikaxsh requested a deployment to integration_tests June 3, 2026 05:46 — with GitHub Actions Waiting

chore: refactor code

71b04e3

vikaxsh requested a deployment to integration_tests June 3, 2026 06:58 — with GitHub Actions Waiting

chore: added a todo

65febe1

vikaxsh requested a deployment to integration_tests June 3, 2026 07:40 — with GitHub Actions Waiting

chore: need to remove this

fdba7b1

vikaxsh requested a deployment to integration_tests June 4, 2026 18:02 — with GitHub Actions Waiting

Merge branch 'staging' into feat/unified-jvm

51fad31

hash-data requested a deployment to integration_tests June 5, 2026 06:16 — with GitHub Actions Waiting

hash-data reviewed Jun 10, 2026

View reviewed changes

vikaxsh added 5 commits June 11, 2026 15:25

feat: Initialize JVM at startup and simplify port allocation

b9265dd

feat: remove upsert, identifierfield from each request set once at st…

311eded

…art per thread session

feat: handle table already exist exception

57b9519

feat: send upsert, table and namespace once in arrow writer requests

c70908a

Merge branch 'staging' of https://github.com/datazip-inc/olake into f…

8caa4da

…eat/unified-jvm

vikaxsh temporarily deployed to integration_tests June 11, 2026 13:59 — with GitHub Actions Inactive

vikaxsh had a problem deploying to integration_tests June 11, 2026 13:59 — with GitHub Actions Error

hash-data reviewed Jun 17, 2026

View reviewed changes

feat: own destination JVM + config lifecycle in WriterPool

824785e

vikaxsh requested a deployment to integration_tests June 17, 2026 09:49 — with GitHub Actions Waiting

Merge branch 'staging' into feat/unified-jvm

269f06b

vikaxsh requested a deployment to integration_tests June 17, 2026 09:54 — with GitHub Actions Waiting

vikaxsh added 2 commits June 17, 2026 15:33

fix: lint issue

b686551

Merge branch 'feat/unified-jvm' of https://github.com/datazip-inc/olake…

e0e78f1

… into feat/unified-jvm

vikaxsh requested a deployment to integration_tests June 17, 2026 10:04 — with GitHub Actions Waiting

fix: remove duplicated Shutdownable

1174788

vikaxsh requested a deployment to integration_tests June 17, 2026 10:36 — with GitHub Actions Waiting

Conversation

vikaxsh commented May 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Changes

Benefits

Type of change

How Has This Been Tested?

Screenshots or Recordings

Documentation

Related PR's (If Any):

Uh oh!

hash-data left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

vikaxsh commented May 26, 2026 •

edited

Loading