apache / apache/seatunnel

[Design Proposal][Zeta] Decouple Hazelcast IMap via StateStore Abstraction

Open
#10,209 6 comments 0 reactions 0 assignees View on GitHub
design Zeta
Dominant language
Java
Stars
9.7k
Forks
2.4k
Avg merge
3d 9h
Merged PRs (30d)
204

Description

### Search before asking

- [x] I had searched in the [feature](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22Feature%22) and found no similar feature requirement.

### Description

## Background

Currently, SeaTunnel relies directly on Hazelcast IMap for internal state management. This tight coupling makes it difficult to:

- Replace the underlying storage engine (e.g., switching from Hazelcast IMap to RocksDB)
- Clearly define the responsibilities of state management components

## Proposal

Introduce a **`StateStore` abstraction** to encapsulate all operations currently performed on Hazelcast IMap, along with a **`StateStoreFactory`** to create instances of `StateStore`.

### Components

- **StateStore**: Abstracts all methods of Hazelcast IMap used internally in SeaTunnel.
- **StateStoreFactory**: Responsible for creating `StateStore` instances.
- **DistributedStoreManager**: Creates and provides the appropriate `StateStore` via the `NodeEngine`.
- As a result, each service no longer interacts with IMap directly. Instead, it passes the `NodeEngine` to the `DistributedStoreManager` to obtain a `StateStore`.

This design allows future replacement of Hazelcast IMap with RocksDB or any other storage engine by simply implementing `StateStore` and `StateStoreFactory` for the new engine. Additionally, creating a storage implementation using RocksDB (or others) becomes clear and straightforward.

### Example

```java
import com.hazelcast.map.IMap;
import com.hazelcast.spi.impl.NodeEngineImpl;

public class IMapFactory implements MapFactory {
private final NodeEngineImpl nodeEngine;

public IMapFactory(NodeEngineImpl nodeEngine) {
this.nodeEngine = nodeEngine;
}

@Override
public MapStorage getMap(String mapName) {
IMap iMap = nodeEngine.getHazelcastInstance().getMap(mapName);
return new IMapStorage<>(iMap);
}
}
```

```java
import com.hazelcast.map.IMap;
import com.hazelcast.map.listener.MapListener;

import javax.annotation.Nonnull;

import java.util.Collection;
import java.util.EventListener;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Function;

public class IMapStorage implements MapStorage {
private final IMap iMap;

public IMapStorage(IMap iMap) {
this.iMap = iMap;
}

@Override
public V putIfAbsent(K key, V value) {
return iMap.putIfAbsent(key, value);
}

@Override
public V compute(K key, BiFunction remappingFunction) {
return iMap.compute(key, remappingFunction);
}

@Override
public void remove(Object key) {
iMap.remove(key);
}

@Override
public V get(Object key) {
return iMap.get(key);
}

@Override
public void put(K key, V value) {
iMap.put(key, value);
}

@Override
public void set(K key, V value) {
iMap.put(key, value);
}

@Override
public Set> entrySet() {
return iMap.entrySet();
}

@Override
public void forEach(BiConsumer action) {
iMap.forEach(action);
}

@Override
public UUID addEntryListener(@Nonnull EventListener listener, boolean includeValue) {
return iMap.addEntryListener((MapListener) listener, includeValue);
}

@Override
public Collection values() {
return iMap.values();
}

@Override
public V getOrDefault(Object key, V defaultValue) {
return iMap.getOrDefault(key, defaultValue);
}

@Override
public V computeIfAbsent(@Nonnull K key, @Nonnull Function func) {
return iMap.computeIfAbsent(key, func);
}

@Override
public V put(@Nonnull K key, @Nonnull V value, long ttl, @Nonnull TimeUnit timeUnit) {
return iMap.put(key, value, ttl, timeUnit);
}

@Override
public boolean isEmpty() {
return iMap.isEmpty();
}

@Override
public boolean containsKey(@Nonnull Object key) {
return iMap.containsKey(key);
}

@Override
public int size() {
return iMap.size();
}
}

```
Feedback is welcome, as there might be parts that need improvement. Thanks!

### Usage Scenario

_No response_

### Related issues

#10181

### Are you willing to submit a PR?

- [x] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)

Contributor guide

No contributing guide indexed for this repository

Research direction

Start by tracing the services that currently interact directly with Hazelcast IMap, then read the proposed StateStore, StateStoreFactory, and DistributedStoreManager responsibilities around the NodeEngine. Done means internal services obtain storage through DistributedStoreManager and the abstraction covers the required IMap operations without direct service-level IMap usage.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
backend, distributed-systems
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
42/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.