diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/processor/StreamProcessor.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/processor/StreamProcessor.java
index 1a718093d..bee8b8988 100644
--- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/processor/StreamProcessor.java
+++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/core/processor/StreamProcessor.java
@@ -14,6 +14,7 @@ import org.slf4j.LoggerFactory;
* @param Type of the specific operator class.
*/
public abstract class StreamProcessor implements Processor {
+
private static final Logger LOGGER = LoggerFactory.getLogger(StreamProcessor.class);
protected List collectors;
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/context/RayRuntimeContext.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/context/RayRuntimeContext.java
index 4cb107cd0..345f67a38 100644
--- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/context/RayRuntimeContext.java
+++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/context/RayRuntimeContext.java
@@ -2,19 +2,39 @@ package io.ray.streaming.runtime.worker.context;
import static io.ray.streaming.util.Config.STREAMING_BATCH_MAX_COUNT;
+import com.google.common.base.Preconditions;
import io.ray.streaming.api.context.RuntimeContext;
import io.ray.streaming.runtime.core.graph.ExecutionTask;
+import io.ray.streaming.state.backend.AbstractKeyStateBackend;
+import io.ray.streaming.state.backend.KeyStateBackend;
+import io.ray.streaming.state.backend.OperatorStateBackend;
+import io.ray.streaming.state.keystate.desc.AbstractStateDescriptor;
+import io.ray.streaming.state.keystate.desc.ListStateDescriptor;
+import io.ray.streaming.state.keystate.desc.MapStateDescriptor;
+import io.ray.streaming.state.keystate.desc.ValueStateDescriptor;
+import io.ray.streaming.state.keystate.state.ListState;
+import io.ray.streaming.state.keystate.state.MapState;
+import io.ray.streaming.state.keystate.state.ValueState;
import java.util.Map;
/**
* Use Ray to implement RuntimeContext.
*/
public class RayRuntimeContext implements RuntimeContext {
+
+ private final Long maxBatch;
+ /**
+ * Backend for keyed state. This might be empty if we're not on a keyed stream.
+ */
+ protected transient KeyStateBackend keyStateBackend;
+ /**
+ * Backend for operator state. This might be empty
+ */
+ protected transient OperatorStateBackend operatorStateBackend;
private int taskId;
private int taskIndex;
private int parallelism;
- private Long batchId;
- private final Long maxBatch;
+ private Long checkpointId;
private Map config;
public RayRuntimeContext(ExecutionTask executionTask, Map config,
@@ -46,8 +66,19 @@ public class RayRuntimeContext implements RuntimeContext {
}
@Override
- public Long getBatchId() {
- return batchId;
+ public Long getCheckpointId() {
+ return checkpointId;
+ }
+
+ @Override
+ public void setCheckpointId(long checkpointId) {
+ if (this.keyStateBackend != null) {
+ this.keyStateBackend.setCheckpointId(checkpointId);
+ }
+ if (this.operatorStateBackend != null) {
+ this.operatorStateBackend.setCheckpointId(checkpointId);
+ }
+ this.checkpointId = checkpointId;
}
@Override
@@ -55,7 +86,48 @@ public class RayRuntimeContext implements RuntimeContext {
return maxBatch;
}
- public void setBatchId(Long batchId) {
- this.batchId = batchId;
+ @Override
+ public Map getConfig() {
+ return config;
+ }
+
+
+ @Override
+ public void setCurrentKey(Object key) {
+ this.keyStateBackend.setCurrentKey(key);
+ }
+
+ @Override
+ public KeyStateBackend getKeyStateBackend() {
+ return keyStateBackend;
+ }
+
+ @Override
+ public void setKeyStateBackend(KeyStateBackend keyStateBackend) {
+ this.keyStateBackend = keyStateBackend;
+ }
+
+ @Override
+ public ValueState getValueState(ValueStateDescriptor stateDescriptor) {
+ stateSanityCheck(stateDescriptor, this.keyStateBackend);
+ return this.keyStateBackend.getValueState(stateDescriptor);
+ }
+
+ @Override
+ public ListState getListState(ListStateDescriptor stateDescriptor) {
+ stateSanityCheck(stateDescriptor, this.keyStateBackend);
+ return this.keyStateBackend.getListState(stateDescriptor);
+ }
+
+ @Override
+ public MapState getMapState(MapStateDescriptor stateDescriptor) {
+ stateSanityCheck(stateDescriptor, this.keyStateBackend);
+ return this.keyStateBackend.getMapState(stateDescriptor);
+ }
+
+ protected void stateSanityCheck(AbstractStateDescriptor stateDescriptor,
+ AbstractKeyStateBackend backend) {
+ Preconditions.checkNotNull(stateDescriptor, "The state properties must not be null");
+ Preconditions.checkNotNull(backend, "backend must not be null");
}
}
diff --git a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/tasks/StreamTask.java b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/tasks/StreamTask.java
index 7bff5101e..d16cc029d 100644
--- a/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/tasks/StreamTask.java
+++ b/streaming/java/streaming-runtime/src/main/java/io/ray/streaming/runtime/worker/tasks/StreamTask.java
@@ -26,6 +26,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public abstract class StreamTask implements Runnable {
+
private static final Logger LOG = LoggerFactory.getLogger(StreamTask.class);
protected int taskId;
@@ -41,8 +42,8 @@ public abstract class StreamTask implements Runnable {
this.worker = worker;
prepareTask();
- this.thread = new Thread(Ray.wrapRunnable(this), this.getClass().getName()
- + "-" + System.currentTimeMillis());
+ this.thread = new Thread(Ray.wrapRunnable(this),
+ this.getClass().getName() + "-" + System.currentTimeMillis());
this.thread.setDaemon(true);
}
@@ -105,8 +106,8 @@ public abstract class StreamTask implements Runnable {
reader = new DataReader(channelIDs, inputActors, queueConf);
}
- RuntimeContext runtimeContext = new RayRuntimeContext(
- worker.getExecutionTask(), worker.getConfig(), executionNode.getParallelism());
+ RuntimeContext runtimeContext = new RayRuntimeContext(worker.getExecutionTask(),
+ worker.getConfig(), executionNode.getParallelism());
processor.open(collectors, runtimeContext);
diff --git a/streaming/java/streaming-state/pom.xml b/streaming/java/streaming-state/pom.xml
new file mode 100644
index 000000000..77e3de57b
--- /dev/null
+++ b/streaming/java/streaming-state/pom.xml
@@ -0,0 +1,95 @@
+
+
+
+ ray-streaming
+ io.ray
+ 0.1-SNAPSHOT
+
+ 4.0.0
+
+ org.ray
+ streaming-state
+
+
+
+
+ org.slf4j
+ slf4j-api
+ ${slf4j.version}
+
+
+ org.slf4j
+ slf4j-log4j12
+ ${slf4j.version}
+
+
+ log4j
+ log4j
+ ${log4j.version}
+
+
+ org.testng
+ testng
+ ${testng.version}
+ test
+
+
+ org.mockito
+ mockito-all
+ ${mockito.version}
+ test
+
+
+
+ com.google.guava
+ guava
+ ${guava.version}
+
+
+
+
+
+
+ org.slf4j
+ slf4j-api
+
+
+ org.slf4j
+ slf4j-log4j12
+
+
+ log4j
+ log4j
+
+
+ org.testng
+ testng
+
+
+ org.mockito
+ mockito-all
+
+
+ de.ruedigermoeller
+ fst
+ ${fst.version}
+
+
+ com.google.guava
+ guava
+
+
+ org.apache.commons
+ commons-lang3
+ 3.3.2
+
+
+ com.beust
+ jcommander
+ 1.27
+
+
+
+
\ No newline at end of file
diff --git a/streaming/java/streaming-state/pom_template.xml b/streaming/java/streaming-state/pom_template.xml
new file mode 100644
index 000000000..adaf7908e
--- /dev/null
+++ b/streaming/java/streaming-state/pom_template.xml
@@ -0,0 +1,22 @@
+
+ {auto_gen_header}
+
+
+ ray-streaming
+ io.ray
+ 0.1-SNAPSHOT
+
+ 4.0.0
+
+ streaming-state
+ ray streaming state
+ ray streaming state
+ jar
+
+
+ {generated_bzl_deps}
+
+
+
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/KeyValueState.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/KeyValueState.java
new file mode 100644
index 000000000..b62571e59
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/KeyValueState.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state;
+
+/**
+ * Key Value State interface.
+ */
+public interface KeyValueState {
+
+ /**
+ * get value from state
+ */
+ V get(K key);
+
+ /**
+ * put key and value into state
+ */
+ void put(K k, V v);
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/PartitionRecord.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/PartitionRecord.java
new file mode 100644
index 000000000..4f58ba7c1
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/PartitionRecord.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state;
+
+import java.io.Serializable;
+
+/**
+ * value record for partition.
+ */
+public class PartitionRecord implements Serializable {
+
+ /**
+ * The partition number of the partitioned value.
+ */
+ private int partitionID;
+ private T value;
+
+ public PartitionRecord() {
+ }
+
+ public PartitionRecord(int partitionID, T value) {
+ this.partitionID = partitionID;
+ this.value = value;
+ }
+
+ public T getValue() {
+ return value;
+ }
+
+ public int getPartitionID() {
+ return partitionID;
+ }
+
+ public void setPartitionID(int partitionID) {
+ this.partitionID = partitionID;
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/StateException.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/StateException.java
new file mode 100644
index 000000000..af608eac5
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/StateException.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state;
+
+/**
+ * RuntimeException wrapper, indicating the exceptions occurs in states.
+ */
+public class StateException extends RuntimeException {
+
+ public StateException(Throwable t) {
+ super(t);
+ }
+
+ public StateException(String msg) {
+ super(msg);
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/StateStoreManager.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/StateStoreManager.java
new file mode 100644
index 000000000..5d55273a8
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/StateStoreManager.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state;
+
+/**
+ * TransactionState interface.
+ *
+ * Streaming State should implement transaction in case of failure,
+ * which in our case is four default method, finish, commit, ackCommit, rollback.
+ */
+public interface StateStoreManager {
+
+ /**
+ * The finish method is used when the batched data is all saved in state.
+ * Normally, serialization job is done here.
+ */
+ void finish(long checkpointId);
+
+ /**
+ * The commit method is used for persistent, and can be used in another thread to reach async
+ * state commit.
+ * Normally, data persistent is done here.
+ */
+ void commit(long checkpointId);
+
+ /**
+ * The ackCommit method is used for cleaning the last checkpoint, and must be called after commit
+ * in the same thread.
+ */
+ void ackCommit(long checkpointId, long timeStamp);
+
+ /**
+ * The rollback method is used for recovering the checkpoint.
+ */
+ void rollBack(long checkpointId);
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/StorageRecord.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/StorageRecord.java
new file mode 100644
index 000000000..03421e679
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/StorageRecord.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state;
+
+import java.io.Serializable;
+
+/**
+ * This Class contains a record with some checkpointId.
+ */
+public class StorageRecord implements Serializable {
+
+ private long checkpointId;
+ private T value;
+
+ public StorageRecord() {
+ }
+
+ public StorageRecord(long checkpointId, T value) {
+ this.checkpointId = checkpointId;
+ this.value = value;
+ }
+
+ public T getValue() {
+ return value;
+ }
+
+ public long getCheckpointId() {
+ return checkpointId;
+ }
+
+ public void setCheckpointId(long checkpointId) {
+ this.checkpointId = checkpointId;
+ }
+
+ @Override
+ public String toString() {
+ if (value != null) {
+ return "checkpointId:" + checkpointId + ", value:" + value;
+ } else {
+ return "checkpointId:" + checkpointId + ", value:null";
+ }
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/AbstractKeyStateBackend.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/AbstractKeyStateBackend.java
new file mode 100644
index 000000000..6edc312c1
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/AbstractKeyStateBackend.java
@@ -0,0 +1,211 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.backend;
+
+import io.ray.streaming.state.StateStoreManager;
+import io.ray.streaming.state.keystate.desc.AbstractStateDescriptor;
+import io.ray.streaming.state.keystate.desc.AbstractStateDescriptor.StateType;
+import io.ray.streaming.state.keystate.state.proxy.ListStateStoreManagerProxy;
+import io.ray.streaming.state.keystate.state.proxy.MapStateStoreManagerProxy;
+import io.ray.streaming.state.keystate.state.proxy.ValueStateStoreManagerProxy;
+import io.ray.streaming.state.store.KeyMapStore;
+import io.ray.streaming.state.store.KeyValueStore;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Transaction support primitive operations like finish, commit, ackcommit and rollback.
+ *
+ * State value modification is not thread safe! By default, every processing thread has its own
+ * space to handle state.
+ */
+public abstract class AbstractKeyStateBackend implements StateStoreManager {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractKeyStateBackend.class);
+
+ protected long currentCheckpointId;
+ protected Object currentKey;
+ protected int keyGroupIndex = -1;
+ protected Map valueManagerProxyHashMap = new HashMap<>();
+ protected Map listManagerProxyHashMap = new HashMap<>();
+ protected Map mapManagerProxyHashMap = new HashMap<>();
+ protected Set descNamespace;
+
+ /**
+ * tablename, KeyValueStore key, checkpointId, content
+ */
+ protected Map>> backStorageCache;
+ private AbstractStateBackend backend;
+
+ public AbstractKeyStateBackend(AbstractStateBackend backend) {
+ this.backStorageCache = new HashMap<>();
+ this.backend = backend;
+ this.descNamespace = new HashSet<>();
+ }
+
+ public void put(AbstractStateDescriptor descriptor, K key, T value) {
+ String desc = descriptor.getIdentify();
+ if (descriptor.getStateType() == StateType.VALUE) {
+ if (this.valueManagerProxyHashMap.containsKey(desc)) {
+ valueManagerProxyHashMap.get(desc).put((String) key, value);
+ }
+ } else if (descriptor.getStateType() == StateType.LIST) {
+ if (this.listManagerProxyHashMap.containsKey(desc)) {
+ listManagerProxyHashMap.get(desc).put((String) key, value);
+ }
+ } else if (descriptor.getStateType() == StateType.MAP) {
+ if (this.mapManagerProxyHashMap.containsKey(desc)) {
+ mapManagerProxyHashMap.get(desc).put((String) key, value);
+ }
+ }
+ }
+
+ public T get(AbstractStateDescriptor descriptor, K key) {
+ String desc = descriptor.getIdentify();
+ if (descriptor.getStateType() == StateType.VALUE) {
+ if (this.valueManagerProxyHashMap.containsKey(desc)) {
+ return (T) valueManagerProxyHashMap.get(desc).get((String) key);
+ }
+ } else if (descriptor.getStateType() == StateType.LIST) {
+ if (this.listManagerProxyHashMap.containsKey(desc)) {
+ return (T) listManagerProxyHashMap.get(desc).get((String) key);
+ }
+ } else if (descriptor.getStateType() == StateType.MAP) {
+ if (this.mapManagerProxyHashMap.containsKey(desc)) {
+ return (T) mapManagerProxyHashMap.get(desc).get((String) key);
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public void finish(long checkpointId) {
+ for (Entry entry : valueManagerProxyHashMap.entrySet()) {
+ entry.getValue().finish(checkpointId);
+ }
+ for (Entry entry : listManagerProxyHashMap.entrySet()) {
+ entry.getValue().finish(checkpointId);
+ }
+ for (Entry entry : mapManagerProxyHashMap.entrySet()) {
+ entry.getValue().finish(checkpointId);
+ }
+ }
+
+ @Override
+ public void commit(long checkpointId) {
+ for (Entry entry : valueManagerProxyHashMap.entrySet()) {
+ entry.getValue().commit(checkpointId);
+ }
+ for (Entry entry : listManagerProxyHashMap.entrySet()) {
+ entry.getValue().commit(checkpointId);
+ }
+ for (Entry entry : mapManagerProxyHashMap.entrySet()) {
+ entry.getValue().commit(checkpointId);
+ }
+ }
+
+ @Override
+ public void ackCommit(long checkpointId, long timeStamp) {
+ for (Entry entry : valueManagerProxyHashMap.entrySet()) {
+ entry.getValue().ackCommit(checkpointId, timeStamp);
+ }
+ for (Entry entry : listManagerProxyHashMap.entrySet()) {
+ entry.getValue().ackCommit(checkpointId, timeStamp);
+ }
+ for (Entry entry : mapManagerProxyHashMap.entrySet()) {
+ entry.getValue().ackCommit(checkpointId, timeStamp);
+ }
+ }
+
+ @Override
+ public void rollBack(long checkpointId) {
+ for (Entry entry : valueManagerProxyHashMap.entrySet()) {
+ LOG.warn("backend rollback:{},{}", entry.getKey(), checkpointId);
+ entry.getValue().rollBack(checkpointId);
+ }
+ for (Entry entry : listManagerProxyHashMap.entrySet()) {
+ LOG.warn("backend rollback:{},{}", entry.getKey(), checkpointId);
+ entry.getValue().rollBack(checkpointId);
+ }
+ for (Entry entry : mapManagerProxyHashMap.entrySet()) {
+ LOG.warn("backend rollback:{},{}", entry.getKey(), checkpointId);
+ entry.getValue().rollBack(checkpointId);
+ }
+ }
+
+ public KeyValueStore> getBackStorage(String tableName) {
+ if (this.backStorageCache.containsKey(tableName)) {
+ return this.backStorageCache.get(tableName);
+ } else {
+ KeyMapStore ikvStore = this.backend.getKeyMapStore(tableName);
+ this.backStorageCache.put(tableName, ikvStore);
+ return ikvStore;
+ }
+ }
+
+ public KeyValueStore> getBackStorage(
+ AbstractStateDescriptor stateDescriptor) {
+ String tableName = this.backend.getTableName(stateDescriptor);
+ return getBackStorage(tableName);
+ }
+
+ public StateStrategy getStateStrategy() {
+ return this.backend.getStateStrategy();
+ }
+
+ public BackendType getBackendType() {
+ return this.backend.getBackendType();
+ }
+
+ public Object getCurrentKey() {
+ return this.currentKey;
+ }
+
+ public abstract void setCurrentKey(Object currentKey);
+
+ public long getCheckpointId() {
+ return this.currentCheckpointId;
+ }
+
+ public void setCheckpointId(long checkpointId) {
+ this.currentCheckpointId = checkpointId;
+ }
+
+ public void setContext(long checkpointId, Object currentKey) {
+ setCheckpointId(checkpointId);
+ setCurrentKey(currentKey);
+ }
+
+ public AbstractStateBackend getBackend() {
+ return backend;
+ }
+
+ public int getKeyGroupIndex() {
+ return this.keyGroupIndex;
+ }
+
+ public void setKeyGroupIndex(int keyGroupIndex) {
+ this.keyGroupIndex = keyGroupIndex;
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/AbstractStateBackend.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/AbstractStateBackend.java
new file mode 100644
index 000000000..d01d7a6b1
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/AbstractStateBackend.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.backend;
+
+import static io.ray.streaming.state.config.ConfigKey.DELIMITER;
+
+import io.ray.streaming.state.config.ConfigKey;
+import io.ray.streaming.state.keystate.desc.AbstractStateDescriptor;
+import io.ray.streaming.state.serialization.KeyMapStoreSerializer;
+import io.ray.streaming.state.store.KeyMapStore;
+import io.ray.streaming.state.store.KeyValueStore;
+import java.io.Serializable;
+import java.util.Map;
+
+/**
+ * This class is the abstract class for different kinds of StateBackend.
+ */
+public abstract class AbstractStateBackend implements Serializable {
+
+ protected final Map config;
+ protected final StateStrategy stateStrategy;
+
+ protected final BackendType backendType;
+ protected int keyGroupIndex = -1;
+
+ protected AbstractStateBackend(Map config) {
+ this.stateStrategy = StateStrategy.getEnum(ConfigKey.getStateStrategyEnum(config));
+ this.backendType = BackendType.getEnum(ConfigKey.getBackendType(config));
+ this.config = config;
+ }
+
+ public abstract KeyValueStore getKeyValueStore(String tableName);
+
+ public abstract KeyMapStore getKeyMapStore(String tableName);
+
+ public abstract KeyMapStore getKeyMapStore(
+ String tableName, KeyMapStoreSerializer keyMapStoreSerializer);
+
+ public BackendType getBackendType() {
+ return backendType;
+ }
+
+ public StateStrategy getStateStrategy() {
+ return stateStrategy;
+ }
+
+ public String getTableName(AbstractStateDescriptor stateDescriptor) {
+ return stateDescriptor.getTableName();
+ }
+
+ public String getStateKey(String descName, String currentKey) {
+ return descName + DELIMITER + currentKey;
+ }
+
+ public void setKeyGroupIndex(int keyGroupIndex) {
+ this.keyGroupIndex = keyGroupIndex;
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/BackendType.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/BackendType.java
new file mode 100644
index 000000000..cc616647e
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/BackendType.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.backend;
+
+/**
+ * Backend Types.
+ */
+public enum BackendType {
+ /**
+ * Saving the state values in memory
+ */
+ MEMORY;
+
+ /**
+ * get the enum from input string value, ignoring the case
+ */
+ public static BackendType getEnum(String value) {
+ for (BackendType v : values()) {
+ if (v.name().equalsIgnoreCase(value)) {
+ return v;
+ }
+ }
+ return MEMORY;
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/KeyStateBackend.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/KeyStateBackend.java
new file mode 100644
index 000000000..9e73295ec
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/KeyStateBackend.java
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.backend;
+
+import io.ray.streaming.state.keystate.KeyGroup;
+import io.ray.streaming.state.keystate.KeyGroupAssignment;
+import io.ray.streaming.state.keystate.desc.ListStateDescriptor;
+import io.ray.streaming.state.keystate.desc.MapStateDescriptor;
+import io.ray.streaming.state.keystate.desc.ValueStateDescriptor;
+import io.ray.streaming.state.keystate.state.ListState;
+import io.ray.streaming.state.keystate.state.MapState;
+import io.ray.streaming.state.keystate.state.ValueState;
+import io.ray.streaming.state.keystate.state.proxy.ListStateStoreManagerProxy;
+import io.ray.streaming.state.keystate.state.proxy.MapStateStoreManagerProxy;
+import io.ray.streaming.state.keystate.state.proxy.ValueStateStoreManagerProxy;
+
+/**
+ * key state backend manager, managing different kinds of states in different thread.
+ * This class is not thread safe.
+ */
+public class KeyStateBackend extends AbstractKeyStateBackend {
+
+ protected final int numberOfKeyGroups;
+ protected final KeyGroup keyGroup;
+
+ public KeyStateBackend(int numberOfKeyGroups, KeyGroup keyGroup,
+ AbstractStateBackend abstractStateBackend) {
+ super(abstractStateBackend);
+ this.numberOfKeyGroups = numberOfKeyGroups;
+ this.keyGroup = keyGroup;
+ }
+
+ /**
+ * get new value state proxy
+ */
+ protected ValueStateStoreManagerProxy newValueStateProxy(
+ ValueStateDescriptor stateDescriptor) {
+ return new ValueStateStoreManagerProxy<>(this, stateDescriptor);
+ }
+
+ public ValueState getValueState(ValueStateDescriptor stateDescriptor) {
+ String desc = stateDescriptor.getIdentify();
+ if (valueManagerProxyHashMap.containsKey(desc)) {
+ return valueManagerProxyHashMap.get(desc).getValueState();
+ } else {
+ ValueStateStoreManagerProxy valueStateProxy = newValueStateProxy(stateDescriptor);
+ valueManagerProxyHashMap.put(desc, valueStateProxy);
+ return valueStateProxy.getValueState();
+ }
+ }
+
+ /**
+ * get new list state proxy
+ */
+ protected ListStateStoreManagerProxy newListStateProxy(
+ ListStateDescriptor stateDescriptor) {
+ return new ListStateStoreManagerProxy<>(this, stateDescriptor);
+ }
+
+ public ListState getListState(ListStateDescriptor stateDescriptor) {
+ String desc = stateDescriptor.getIdentify();
+ if (listManagerProxyHashMap.containsKey(desc)) {
+ ListStateStoreManagerProxy listStateProxy = listManagerProxyHashMap.get(desc);
+ return listStateProxy.getListState();
+ } else {
+ ListStateStoreManagerProxy listStateProxy = newListStateProxy(stateDescriptor);
+ listManagerProxyHashMap.put(desc, listStateProxy);
+ return listStateProxy.getListState();
+ }
+ }
+
+ /**
+ * get map state proxy
+ */
+ protected MapStateStoreManagerProxy newMapStateProxy(
+ MapStateDescriptor stateDescriptor) {
+ return new MapStateStoreManagerProxy<>(this, stateDescriptor);
+ }
+
+ public MapState getMapState(MapStateDescriptor stateDescriptor) {
+ String desc = stateDescriptor.getIdentify();
+ if (mapManagerProxyHashMap.containsKey(desc)) {
+ MapStateStoreManagerProxy mapStateProxy = mapManagerProxyHashMap.get(desc);
+ return mapStateProxy.getMapState();
+ } else {
+ MapStateStoreManagerProxy mapStateProxy = newMapStateProxy(stateDescriptor);
+ mapManagerProxyHashMap.put(desc, mapStateProxy);
+ return mapStateProxy.getMapState();
+ }
+ }
+
+ @Override
+ public void setCurrentKey(Object currentKey) {
+ super.keyGroupIndex = KeyGroupAssignment
+ .assignKeyGroupIndexForKey(currentKey, numberOfKeyGroups);
+ super.currentKey = currentKey;
+ }
+
+ public int getNumberOfKeyGroups() {
+ return numberOfKeyGroups;
+ }
+
+ public KeyGroup getKeyGroup() {
+ return keyGroup;
+ }
+
+ public void close() {
+ for (ValueStateStoreManagerProxy proxy : valueManagerProxyHashMap.values()) {
+ proxy.close();
+ }
+ for (ListStateStoreManagerProxy proxy : listManagerProxyHashMap.values()) {
+ proxy.close();
+ }
+ for (MapStateStoreManagerProxy proxy : mapManagerProxyHashMap.values()) {
+ proxy.close();
+ }
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/OperatorStateBackend.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/OperatorStateBackend.java
new file mode 100644
index 000000000..68bdac581
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/OperatorStateBackend.java
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.backend;
+
+import io.ray.streaming.state.keystate.desc.ListStateDescriptor;
+import io.ray.streaming.state.keystate.state.ListState;
+import io.ray.streaming.state.keystate.state.impl.OperatorStateImpl;
+import io.ray.streaming.state.keystate.state.proxy.ListStateStoreManagerProxy;
+
+/**
+ * OperatorState manager for getting split or union list state.
+ */
+public class OperatorStateBackend extends AbstractKeyStateBackend {
+
+ public OperatorStateBackend(AbstractStateBackend backend) {
+ super(backend);
+ }
+
+ @Override
+ public void setCurrentKey(Object currentKey) {
+ super.currentKey = currentKey;
+ }
+
+ protected ListStateStoreManagerProxy newListStateStoreManagerProxy(
+ ListStateDescriptor stateDescriptor) {
+ return new ListStateStoreManagerProxy<>(this, stateDescriptor);
+ }
+
+ /**
+ * get spitted List for different operator instance.
+ */
+ public ListState getSplitListState(ListStateDescriptor stateDescriptor) {
+ String desc = stateDescriptor.getIdentify();
+ if (listManagerProxyHashMap.containsKey(desc)) {
+ ListStateStoreManagerProxy listStateProxy = listManagerProxyHashMap.get(desc);
+ return listStateProxy.getListState();
+ } else {
+ ListStateStoreManagerProxy listStateProxy = newListStateStoreManagerProxy(stateDescriptor);
+ listManagerProxyHashMap.put(desc, listStateProxy);
+ ((OperatorStateImpl) (listStateProxy.getListState())).setSplit(true);
+ return listStateProxy.getListState();
+ }
+ }
+
+ /**
+ * get a union List for different operator instance.
+ */
+ public ListState getUnionListState(ListStateDescriptor stateDescriptor) {
+ String desc = stateDescriptor.getIdentify();
+ if (listManagerProxyHashMap.containsKey(desc)) {
+ ListStateStoreManagerProxy listStateProxy = listManagerProxyHashMap.get(desc);
+ return listStateProxy.getListState();
+ } else {
+ ListStateStoreManagerProxy listStateProxy = newListStateStoreManagerProxy(stateDescriptor);
+ listManagerProxyHashMap.put(desc, listStateProxy);
+ ((OperatorStateImpl) (listStateProxy.getListState())).init();
+ return listStateProxy.getListState();
+ }
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/StateBackendBuilder.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/StateBackendBuilder.java
new file mode 100644
index 000000000..27e849a96
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/StateBackendBuilder.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.backend;
+
+import io.ray.streaming.state.backend.impl.MemoryStateBackend;
+import io.ray.streaming.state.config.ConfigKey;
+import java.util.Map;
+
+/**
+ * State Backend Builder.
+ */
+public class StateBackendBuilder {
+
+ private static AbstractStateBackend getStateBackend(Map config,
+ BackendType type) {
+ switch (type) {
+ case MEMORY:
+ return new MemoryStateBackend(config);
+ default:
+ throw new RuntimeException(type.name() + " not supported");
+ }
+ }
+
+ public static AbstractStateBackend buildStateBackend(Map config) {
+ BackendType type;
+ if (config == null) {
+ type = BackendType.MEMORY;
+ } else {
+ type = BackendType.getEnum(config.get(ConfigKey.STATE_BACKEND_TYPE));
+ }
+
+ return getStateBackend(config, type);
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/StateStrategy.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/StateStrategy.java
new file mode 100644
index 000000000..40d231cf8
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/StateStrategy.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.backend;
+
+/**
+ * This class describe State Saving Model.
+ */
+public enum StateStrategy {
+ /**
+ * save two version together in case of rollback.
+ */
+ DUAL_VERSION,
+
+ /**
+ * for storage supporting mvcc, we save only current version.
+ */
+ SINGLE_VERSION;
+
+
+ public static StateStrategy getEnum(String value) {
+ for (StateStrategy v : values()) {
+ if (v.name().equalsIgnoreCase(value)) {
+ return v;
+ }
+ }
+ throw new IllegalArgumentException(value + " strategy is not supported");
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/impl/MemoryStateBackend.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/impl/MemoryStateBackend.java
new file mode 100644
index 000000000..67eeebf33
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/backend/impl/MemoryStateBackend.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.backend.impl;
+
+import io.ray.streaming.state.backend.AbstractStateBackend;
+import io.ray.streaming.state.serialization.KeyMapStoreSerializer;
+import io.ray.streaming.state.store.KeyMapStore;
+import io.ray.streaming.state.store.KeyValueStore;
+import io.ray.streaming.state.store.impl.MemoryKeyMapStore;
+import io.ray.streaming.state.store.impl.MemoryKeyValueStore;
+import java.util.Map;
+
+/**
+ * MemoryStateBackend. Supporting memory store.
+ */
+public class MemoryStateBackend extends AbstractStateBackend {
+
+ public MemoryStateBackend(Map config) {
+ super(config);
+ }
+
+ @Override
+ public KeyValueStore getKeyValueStore(String tableName) {
+ return new MemoryKeyValueStore<>();
+ }
+
+ @Override
+ public KeyMapStore getKeyMapStore(String tableName) {
+ return new MemoryKeyMapStore<>();
+ }
+
+ @Override
+ public KeyMapStore getKeyMapStore(
+ String tableName, KeyMapStoreSerializer keyMapStoreSerializer) {
+ return new MemoryKeyMapStore<>();
+ }
+
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/config/ConfigHelper.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/config/ConfigHelper.java
new file mode 100644
index 000000000..0fbf88f3e
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/config/ConfigHelper.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.config;
+
+import java.util.Map;
+
+/**
+ * Config Helper figure out the config info.
+ * Todo replace this to config module.
+ */
+public class ConfigHelper {
+
+ public static int getIntegerOrDefault(Map config, String configKey, int defaultValue) {
+ if (config.containsKey(configKey)) {
+ return Integer.valueOf(String.valueOf(config.get(configKey)));
+ } else {
+ return defaultValue;
+ }
+ }
+
+ public static String getStringOrDefault(Map config, String configKey, String defaultValue) {
+ if (config.containsKey(configKey)) {
+ return String.valueOf(config.get(configKey));
+ } else {
+ return defaultValue;
+ }
+ }
+
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/config/ConfigKey.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/config/ConfigKey.java
new file mode 100644
index 000000000..f7d8a42a0
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/config/ConfigKey.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.config;
+
+import java.util.Map;
+
+/**
+ * state config keys.
+ * Todo replace this to config module.
+ */
+public final class ConfigKey {
+
+ /**
+ * backend
+ */
+ public static final String STATE_BACKEND_TYPE = "state.backend.type";
+ public static final String STATE_TABLE_NAME = "state.table.name";
+ public static final String STATE_STRATEGY_MODE = "state.strategy.mode";
+ public static final String NUMBER_PER_CHECKPOINT = "number.per.checkpoint";
+ public static final String JOB_MAX_PARALLEL = "job.max.parallel";
+ public static final String DELIMITER = "\u0001\u0008"; // for String delimiter
+
+ private ConfigKey() {
+ throw new AssertionError();
+ }
+
+ public static String getStateStrategyEnum(Map config) {
+ return ConfigHelper.getStringOrDefault(config, STATE_STRATEGY_MODE, "DUAL_VERSION");
+ }
+
+ public static String getBackendType(Map config) {
+ return ConfigHelper.getStringOrDefault(config, STATE_BACKEND_TYPE, "MEMORY");
+ }
+
+ public static int getNumberPerCheckpoint(Map config) {
+ return ConfigHelper.getIntegerOrDefault(config, NUMBER_PER_CHECKPOINT, 5);
+ }
+
+ public static String getStateTableName(Map config) {
+ return ConfigHelper.getStringOrDefault(config, STATE_TABLE_NAME, "table");
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/KeyGroup.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/KeyGroup.java
new file mode 100644
index 000000000..fb24c72d1
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/KeyGroup.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.keystate;
+
+import com.google.common.base.Preconditions;
+import java.io.Serializable;
+
+/**
+ * This class defines key-groups. Key-groups is
+ * the key space in a job, which is partitioned for keyed state processing in state backend.
+ * The boundaries of the key-group are inclusive.
+ */
+public class KeyGroup implements Serializable {
+
+ private final int startIndex;
+ private final int endIndex;
+
+ /**
+ * Defines the range [startIndex, endIndex]
+ *
+ * @param startIndex start of the range (inclusive)
+ * @param endIndex end of the range (inclusive)
+ */
+ public KeyGroup(int startIndex, int endIndex) {
+ Preconditions.checkArgument(startIndex >= 0 && startIndex <= endIndex);
+ this.startIndex = startIndex;
+ this.endIndex = endIndex;
+ Preconditions.checkArgument(size() >= 0, "overflow detected.");
+ }
+
+ /**
+ * @return The number of key-group in the range
+ */
+ public int size() {
+ return 1 + endIndex - startIndex;
+ }
+
+ public int getStartIndex() {
+ return startIndex;
+ }
+
+ public int getEndIndex() {
+ return endIndex;
+ }
+
+ @Override
+ public String toString() {
+ return "KeyGroup{" + "startIndex=" + startIndex + ", endIndex=" + endIndex + '}';
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/KeyGroupAssignment.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/KeyGroupAssignment.java
new file mode 100644
index 000000000..77b465f74
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/KeyGroupAssignment.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.keystate;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * This class defines key-group assignment algorithm。
+ */
+public final class KeyGroupAssignment {
+
+ /**
+ * Computes the range of key-groups that are assigned for a given operator instance.
+ *
+ * @param maxParallelism Maximal parallelism of the job.
+ * @param parallelism Parallelism for the job. <= maxParallelism.
+ * @param index index of the operator instance.
+ */
+ public static KeyGroup getKeyGroup(int maxParallelism, int parallelism, int index) {
+ Preconditions.checkArgument(maxParallelism >= parallelism,
+ "Maximum parallelism (%s) must not be smaller than parallelism(%s)", maxParallelism,
+ parallelism);
+
+ int start = index == 0 ? 0 : ((index * maxParallelism - 1) / parallelism) + 1;
+ int end = ((index + 1) * maxParallelism - 1) / parallelism;
+ return new KeyGroup(start, end);
+ }
+
+ /**
+ * Assigning the key to a key-group index.
+ *
+ * @param key the key to assign.
+ * @param maxParallelism the maximum parallelism.
+ * @return the key-group index to which the given key is assigned.
+ */
+ public static int assignKeyGroupIndexForKey(Object key, int maxParallelism) {
+ return Math.abs(key.hashCode() % maxParallelism);
+ }
+
+ public static Map> computeKeyGroupToTask(int maxParallelism,
+ List targetTasks) {
+ Map> keyGroupToTask = new ConcurrentHashMap<>();
+ for (int index = 0; index < targetTasks.size(); index++) {
+ KeyGroup taskKeyGroup = getKeyGroup(maxParallelism, targetTasks.size(), index);
+ for (int groupId = taskKeyGroup.getStartIndex(); groupId <= taskKeyGroup.getEndIndex();
+ groupId++) {
+ keyGroupToTask.put(groupId, ImmutableList.of(targetTasks.get(index)));
+ }
+ }
+ return keyGroupToTask;
+ }
+
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/desc/AbstractStateDescriptor.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/desc/AbstractStateDescriptor.java
new file mode 100644
index 000000000..c85decf95
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/desc/AbstractStateDescriptor.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.keystate.desc;
+
+import com.google.common.base.Preconditions;
+import io.ray.streaming.state.keystate.state.State;
+
+/**
+ * This class defines basic data structures of StateDescriptor.
+ */
+public abstract class AbstractStateDescriptor {
+
+ private final String name;
+ private String tableName;
+ private Class type;
+
+ protected AbstractStateDescriptor(String name, Class type) {
+ this.name = name;
+ this.type = type;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public Class getType() {
+ return type;
+ }
+
+ protected Class setType(Class type) {
+ return type;
+ }
+
+ public String getTableName() {
+ return tableName;
+ }
+
+ public void setTableName(String tableName) {
+ this.tableName = tableName;
+ }
+
+ public abstract StateType getStateType();
+
+ public String getIdentify() {
+ Preconditions.checkArgument(this.tableName != null, "table name must not be null.");
+ Preconditions.checkArgument(this.name != null, "table name must not be null.");
+ return this.name;
+ }
+
+ @Override
+ public String toString() {
+ return "AbstractStateDescriptor{" + "tableName='" + tableName + '\'' + ", name='" + name + '\''
+ + ", type=" + type + '}';
+ }
+
+ public enum StateType {
+ /**
+ * value state
+ */
+ VALUE,
+
+ /**
+ * list state
+ */
+ LIST,
+
+ /**
+ * map state
+ */
+ MAP
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/desc/ListStateDescriptor.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/desc/ListStateDescriptor.java
new file mode 100644
index 000000000..c6a7af6b6
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/desc/ListStateDescriptor.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.keystate.desc;
+
+
+import static io.ray.streaming.state.config.ConfigKey.DELIMITER;
+
+import io.ray.streaming.state.keystate.state.ListState;
+
+/**
+ * ListStateDescriptor.
+ */
+public class ListStateDescriptor extends AbstractStateDescriptor, T> {
+
+ private final boolean isOperatorList;
+ private int index;
+ private int partitionNum;
+
+ private ListStateDescriptor(String name, Class type, boolean isOperatorList) {
+ super(name, type);
+ this.isOperatorList = isOperatorList;
+ }
+
+ public static ListStateDescriptor build(String name, Class type) {
+ return build(name, type, false);
+ }
+
+ public static ListStateDescriptor build(String name, Class type,
+ boolean isOperatorList) {
+ return new ListStateDescriptor<>(name, type, isOperatorList);
+ }
+
+ public boolean isOperatorList() {
+ return isOperatorList;
+ }
+
+ public int getIndex() {
+ return index;
+ }
+
+ public void setIndex(int index) {
+ this.index = index;
+ }
+
+ public int getPartitionNumber() {
+ return partitionNum;
+ }
+
+ public void setPartitionNumber(int number) {
+ this.partitionNum = number;
+ }
+
+ @Override
+ public StateType getStateType() {
+ return StateType.LIST;
+ }
+
+ @Override
+ public String getIdentify() {
+ if (isOperatorList) {
+ return String
+ .format("%s%s%d%s%d", super.getIdentify(), DELIMITER, partitionNum, DELIMITER, index);
+ } else {
+ return super.getIdentify();
+ }
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/desc/MapStateDescriptor.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/desc/MapStateDescriptor.java
new file mode 100644
index 000000000..15373265e
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/desc/MapStateDescriptor.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.keystate.desc;
+
+import io.ray.streaming.state.keystate.state.MapState;
+import java.util.Map;
+
+/**
+ * MapStateDescriptor.
+ */
+public class MapStateDescriptor extends AbstractStateDescriptor, Map> {
+
+ public MapStateDescriptor(String name, Class keyType, Class valueType) {
+ super(name, null);
+ // TODO: use the types to help serde
+ }
+
+ public static MapStateDescriptor build(String name, Class keyType,
+ Class valueType) {
+ return new MapStateDescriptor<>(name, keyType, valueType);
+ }
+
+ @Override
+ public StateType getStateType() {
+ return StateType.MAP;
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/desc/ValueStateDescriptor.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/desc/ValueStateDescriptor.java
new file mode 100644
index 000000000..e067ba71d
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/desc/ValueStateDescriptor.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.keystate.desc;
+
+import io.ray.streaming.state.keystate.state.ValueState;
+
+/**
+ * ValueStateDescriptor.
+ */
+public class ValueStateDescriptor extends AbstractStateDescriptor, T> {
+
+ private final T defaultValue;
+
+ public ValueStateDescriptor(String name, Class type, T defaultValue) {
+ super(name, type);
+ this.defaultValue = defaultValue;
+ }
+
+ public static ValueStateDescriptor build(String name, Class type, T defaultValue) {
+ return new ValueStateDescriptor<>(name, type, defaultValue);
+ }
+
+ public T getDefaultValue() {
+ return defaultValue;
+ }
+
+ @Override
+ public StateType getStateType() {
+ return StateType.VALUE;
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/ListState.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/ListState.java
new file mode 100644
index 000000000..d5efa6670
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/ListState.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.keystate.state;
+
+import java.util.List;
+
+/**
+ * ListState interface.
+ */
+public interface ListState extends UnaryState> {
+
+ /**
+ * add the value to list
+ *
+ * @param value the new value
+ */
+ void add(T value);
+
+ /**
+ * update list state
+ *
+ * @param list the new value
+ */
+ void update(List list);
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/MapState.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/MapState.java
new file mode 100644
index 000000000..663da8f71
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/MapState.java
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.keystate.state;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Map.Entry;
+
+/**
+ * MapState interface.
+ */
+public interface MapState extends UnaryState> {
+
+ /**
+ * Returns the current value associated with the given key.
+ *
+ * @param key The key of the mapping
+ * @return The value of the mapping with the given key
+ */
+ V get(K key);
+
+ /**
+ * Associates a new value with the given key.
+ *
+ * @param key The key of the mapping
+ * @param value The new value of the mapping
+ */
+ void put(K key, V value);
+
+ /**
+ * Resets the state value.
+ *
+ * @param map The mappings for reset in this state
+ */
+ void update(Map map);
+
+ /**
+ * Copies all of the mappings from the given map into the state.
+ *
+ * @param map The mappings to be stored in this state
+ */
+ void putAll(Map map);
+
+ /**
+ * Deletes the mapping of the given key.
+ *
+ * @param key The key of the mapping
+ */
+ void remove(K key);
+
+ /**
+ * Returns whether there exists the given mapping.
+ *
+ * @param key The key of the mapping
+ * @return True if there exists a mapping whose key equals to the given key
+ */
+ default boolean contains(K key) {
+ return get().containsKey(key);
+ }
+
+ /**
+ * Returns all the mappings in the state
+ *
+ * @return An iterable view of all the key-value pairs in the state.
+ */
+ default Iterable> entries() {
+ return get().entrySet();
+ }
+
+ /**
+ * Returns all the keys in the state
+ *
+ * @return An iterable view of all the keys in the state.
+ */
+ default Iterable keys() {
+ return get().keySet();
+ }
+
+ /**
+ * Returns all the values in the state.
+ *
+ * @return An iterable view of all the values in the state.
+ */
+ default Iterable values() {
+ return get().values();
+ }
+
+ /**
+ * Iterates over all the mappings in the state.
+ *
+ * @return An iterator over all the mappings in the state
+ */
+ default Iterator> iterator() {
+ return get().entrySet().iterator();
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/State.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/State.java
new file mode 100644
index 000000000..6b18c62d8
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/State.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.keystate.state;
+
+/**
+ * State interface.
+ */
+public interface State {
+
+ /**
+ * set current key of the state
+ */
+ void setCurrentKey(Object currentKey);
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/UnaryState.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/UnaryState.java
new file mode 100644
index 000000000..a67196d20
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/UnaryState.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.keystate.state;
+
+/**
+ * one value per state.
+ */
+public interface UnaryState extends State {
+
+ /**
+ * get the value in state
+ *
+ * @return the value in state
+ */
+ O get();
+}
\ No newline at end of file
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/ValueState.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/ValueState.java
new file mode 100644
index 000000000..e26b23b8a
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/ValueState.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.keystate.state;
+
+/**
+ * ValueState interface.
+ */
+public interface ValueState extends UnaryState {
+
+ /**
+ * update the value
+ *
+ * @param value the new value
+ */
+ void update(T value);
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/impl/ListStateImpl.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/impl/ListStateImpl.java
new file mode 100644
index 000000000..a56da03ec
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/impl/ListStateImpl.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.keystate.state.impl;
+
+import io.ray.streaming.state.backend.AbstractKeyStateBackend;
+import io.ray.streaming.state.keystate.desc.ListStateDescriptor;
+import io.ray.streaming.state.keystate.state.ListState;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * ListState implementation.
+ */
+public class ListStateImpl implements ListState {
+
+ private final StateHelper> helper;
+
+ public ListStateImpl(ListStateDescriptor descriptor, AbstractKeyStateBackend backend) {
+ this.helper = new StateHelper<>(backend, descriptor);
+ }
+
+ @Override
+ public List get() {
+ List list = helper.get();
+ if (list == null) {
+ list = new ArrayList<>();
+ }
+ return list;
+ }
+
+ @Override
+ public void add(V value) {
+ List list = helper.get();
+ if (list == null) {
+ list = new ArrayList<>();
+ }
+ list.add(value);
+ helper.put(list);
+ }
+
+ @Override
+ public void update(List list) {
+ if (list == null) {
+ list = new ArrayList<>();
+ }
+ helper.put(list);
+ }
+
+ @Override
+ public void setCurrentKey(Object currentKey) {
+ helper.setCurrentKey(currentKey);
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/impl/MapStateImpl.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/impl/MapStateImpl.java
new file mode 100644
index 000000000..0d0729133
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/impl/MapStateImpl.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.keystate.state.impl;
+
+import io.ray.streaming.state.backend.KeyStateBackend;
+import io.ray.streaming.state.keystate.desc.MapStateDescriptor;
+import io.ray.streaming.state.keystate.state.MapState;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * MapState implementation.
+ */
+public class MapStateImpl implements MapState {
+
+ private final StateHelper> helper;
+
+ public MapStateImpl(MapStateDescriptor descriptor, KeyStateBackend backend) {
+ this.helper = new StateHelper<>(backend, descriptor);
+ }
+
+ @Override
+ public Map get() {
+ Map map = helper.get();
+ if (map == null) {
+ map = new HashMap<>();
+ }
+ return map;
+ }
+
+ @Override
+ public V get(K key) {
+ Map map = get();
+ return map.get(key);
+ }
+
+ @Override
+ public void put(K key, V value) {
+ Map map = get();
+
+ map.put(key, value);
+ helper.put(map);
+ }
+
+ @Override
+ public void update(Map map) {
+ if (map == null) {
+ map = new HashMap<>();
+ }
+ helper.put(map);
+ }
+
+ @Override
+ public void putAll(Map newMap) {
+ Map map = get();
+
+ map.putAll(newMap);
+ helper.put(map);
+ }
+
+ @Override
+ public void remove(K key) {
+ Map map = get();
+
+ map.remove(key);
+ helper.put(map);
+ }
+
+ /**
+ * set current key of the state
+ */
+ @Override
+ public void setCurrentKey(Object currentKey) {
+ helper.setCurrentKey(currentKey);
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/impl/OperatorStateImpl.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/impl/OperatorStateImpl.java
new file mode 100644
index 000000000..e1b8defd3
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/impl/OperatorStateImpl.java
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.keystate.state.impl;
+
+import static io.ray.streaming.state.config.ConfigKey.DELIMITER;
+
+import com.google.common.base.Preconditions;
+import io.ray.streaming.state.PartitionRecord;
+import io.ray.streaming.state.backend.AbstractKeyStateBackend;
+import io.ray.streaming.state.keystate.desc.ListStateDescriptor;
+import io.ray.streaming.state.keystate.state.ListState;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * This class defines the implementation of operator state.
+ * When the state is initialized, we must scan the whole table.
+ * And if the state type is splitList, all the records must be spitted.
+ */
+public class OperatorStateImpl implements ListState {
+
+ private final ListStateDescriptor descriptor;
+ private StateHelper>> helper;
+ private List> allList;
+
+ private AtomicBoolean hasInit;
+ private boolean isSplit;
+
+ public OperatorStateImpl(ListStateDescriptor descriptor, AbstractKeyStateBackend backend) {
+ this.descriptor = descriptor;
+ this.helper = new StateHelper<>(backend, descriptor);
+ this.isSplit = false;
+ this.hasInit = new AtomicBoolean(false);
+ this.allList = new ArrayList<>();
+ }
+
+ private void splitList() {
+ // fetch target list and save
+ List> list = new ArrayList<>();
+ int step = descriptor.getPartitionNumber();
+ Preconditions.checkArgument(step > 0);
+
+ for (int round = 0; round * step <= allList.size(); round++) {
+ int m = round * step + descriptor.getIndex();
+ if (m < allList.size()) {
+ PartitionRecord tmp = allList.get(m);
+ tmp.setPartitionID(descriptor.getPartitionNumber());
+ list.add(tmp);
+ }
+ }
+ helper.put(list, getStateKey());
+ allList.clear();
+ }
+
+ private void scan() {
+ int partitionNum = -1;
+ int index = 0;
+ while (true) {
+ List> list = helper.getBackend()
+ .get(descriptor, getKey(descriptor.getIdentify(), index));
+ if (list != null && !list.isEmpty()) {
+ partitionNum = list.get(0).getPartitionID();
+ allList.addAll(list);
+ }
+ if (++index >= partitionNum) {
+ break;
+ }
+ }
+ }
+
+ public void init() {
+ scan();
+
+ if (isSplit) {
+ splitList();
+ }
+ }
+
+ private String getKey(String descName, int index) {
+ String[] stringList = descName.split(DELIMITER);
+ return String.format("%s%s%s%s%d", stringList[0], DELIMITER, stringList[1], DELIMITER, index);
+ }
+
+ protected String getStateKey() {
+ return getKey(this.descriptor.getIdentify(), this.descriptor.getIndex());
+ }
+
+ @Override
+ public void setCurrentKey(Object currentKey) {
+ throw new UnsupportedOperationException("OperatorState cannot set current key");
+ }
+
+ @Override
+ public List get() {
+ if (!hasInit.getAndSet(true)) {
+ init();
+ }
+ List> prList = helper.get(getStateKey());
+ List list = new ArrayList<>();
+ for (PartitionRecord pr : prList) {
+ list.add(pr.getValue());
+ }
+ return list;
+ }
+
+ @Override
+ public void add(V value) {
+ if (!hasInit.getAndSet(true)) {
+ init();
+ }
+ List> list = helper.get(getStateKey());
+ if (list == null) {
+ list = new ArrayList<>();
+ }
+ list.add(new PartitionRecord<>(descriptor.getPartitionNumber(), value));
+ helper.put(list, getStateKey());
+ }
+
+ @Override
+ public void update(List list) {
+ List> prList = new ArrayList<>();
+ if (list != null) {
+ for (V value : list) {
+ prList.add(new PartitionRecord<>(descriptor.getPartitionNumber(), value));
+ }
+ }
+ helper.put(prList);
+ }
+
+ public void setSplit(boolean split) {
+ this.isSplit = split;
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/impl/StateHelper.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/impl/StateHelper.java
new file mode 100644
index 000000000..23511081a
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/impl/StateHelper.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.keystate.state.impl;
+
+import com.google.common.base.Preconditions;
+import io.ray.streaming.state.backend.AbstractKeyStateBackend;
+import io.ray.streaming.state.keystate.desc.AbstractStateDescriptor;
+
+/**
+ * State Helper Class.
+ */
+public class StateHelper {
+
+ private final AbstractKeyStateBackend backend;
+ private final AbstractStateDescriptor descriptor;
+
+ public StateHelper(AbstractKeyStateBackend backend, AbstractStateDescriptor descriptor) {
+ this.backend = backend;
+ this.descriptor = descriptor;
+ }
+
+ protected String getStateKey(String descName) {
+ Preconditions.checkNotNull(backend, "KeyedBackend must not be null");
+ Preconditions.checkNotNull(backend.getCurrentKey(), "currentKey must not be null");
+ return this.backend.getBackend().getStateKey(descName, backend.getCurrentKey().toString());
+ }
+
+ public void put(T value, String key) {
+ backend.put(descriptor, key, value);
+ }
+
+ public void put(T value) {
+ put(value, getStateKey(descriptor.getIdentify()));
+ }
+
+ public T get(String key) {
+ return backend.get(descriptor, key);
+ }
+
+ public T get() {
+ return get(getStateKey(descriptor.getIdentify()));
+ }
+
+ public void setCurrentKey(Object currentKey) {
+ Preconditions.checkNotNull(backend, "KeyedBackend must not be null");
+ this.backend.setCurrentKey(currentKey);
+ }
+
+ public void setKeyGroupIndex(int keyGroupIndex) {
+ this.backend.setKeyGroupIndex(keyGroupIndex);
+ }
+
+ public void resetKeyGroupIndex() {
+ this.backend.setKeyGroupIndex(-1);
+ }
+
+ public AbstractStateDescriptor getDescriptor() {
+ return descriptor;
+ }
+
+ public AbstractKeyStateBackend getBackend() {
+ return backend;
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/impl/ValueStateImpl.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/impl/ValueStateImpl.java
new file mode 100644
index 000000000..50a72ec65
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/impl/ValueStateImpl.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.keystate.state.impl;
+
+import io.ray.streaming.state.backend.KeyStateBackend;
+import io.ray.streaming.state.keystate.desc.ValueStateDescriptor;
+import io.ray.streaming.state.keystate.state.ValueState;
+
+/**
+ * ValueState implementation.
+ */
+public class ValueStateImpl implements ValueState {
+
+ private final StateHelper helper;
+
+ public ValueStateImpl(ValueStateDescriptor descriptor, KeyStateBackend backend) {
+ this.helper = new StateHelper<>(backend, descriptor);
+ }
+
+ @Override
+ public void update(T value) {
+ helper.put(value);
+ }
+
+ @Override
+ public T get() {
+ T value = helper.get();
+ if (null == value) {
+ return ((ValueStateDescriptor) helper.getDescriptor()).getDefaultValue();
+ } else {
+ return value;
+ }
+ }
+
+ /**
+ * set current key of the state
+ */
+ @Override
+ public void setCurrentKey(Object currentKey) {
+ helper.setCurrentKey(currentKey);
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/proxy/ListStateStoreManagerProxy.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/proxy/ListStateStoreManagerProxy.java
new file mode 100644
index 000000000..9f25ec4f4
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/proxy/ListStateStoreManagerProxy.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.keystate.state.proxy;
+
+import io.ray.streaming.state.KeyValueState;
+import io.ray.streaming.state.backend.AbstractKeyStateBackend;
+import io.ray.streaming.state.keystate.desc.ListStateDescriptor;
+import io.ray.streaming.state.keystate.state.ListState;
+import io.ray.streaming.state.keystate.state.impl.ListStateImpl;
+import io.ray.streaming.state.keystate.state.impl.OperatorStateImpl;
+import io.ray.streaming.state.strategy.StateStoreManagerProxy;
+import java.util.List;
+
+/**
+ * This class defines ListState Wrapper, connecting state and backend.
+ */
+public class ListStateStoreManagerProxy extends StateStoreManagerProxy> implements
+ KeyValueState> {
+
+ private final ListState listState;
+
+ public ListStateStoreManagerProxy(AbstractKeyStateBackend keyStateBackend,
+ ListStateDescriptor stateDescriptor) {
+ super(keyStateBackend, stateDescriptor);
+ if (stateDescriptor.isOperatorList()) {
+ this.listState = new OperatorStateImpl<>(stateDescriptor, keyStateBackend);
+ } else {
+ this.listState = new ListStateImpl<>(stateDescriptor, keyStateBackend);
+ }
+ }
+
+ public ListState getListState() {
+ return this.listState;
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/proxy/MapStateStoreManagerProxy.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/proxy/MapStateStoreManagerProxy.java
new file mode 100644
index 000000000..a30e5084a
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/proxy/MapStateStoreManagerProxy.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.keystate.state.proxy;
+
+import io.ray.streaming.state.KeyValueState;
+import io.ray.streaming.state.backend.KeyStateBackend;
+import io.ray.streaming.state.keystate.desc.MapStateDescriptor;
+import io.ray.streaming.state.keystate.state.MapState;
+import io.ray.streaming.state.keystate.state.impl.MapStateImpl;
+import io.ray.streaming.state.strategy.StateStoreManagerProxy;
+import java.util.Map;
+
+/**
+ * This class defines MapState Wrapper, connecting state and backend.
+ */
+public class MapStateStoreManagerProxy extends StateStoreManagerProxy> implements
+ KeyValueState> {
+
+ private final MapStateImpl mapState;
+
+ public MapStateStoreManagerProxy(KeyStateBackend keyStateBackend,
+ MapStateDescriptor stateDescriptor) {
+ super(keyStateBackend, stateDescriptor);
+ this.mapState = new MapStateImpl<>(stateDescriptor, keyStateBackend);
+ }
+
+ public MapState getMapState() {
+ return this.mapState;
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/proxy/ValueStateStoreManagerProxy.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/proxy/ValueStateStoreManagerProxy.java
new file mode 100644
index 000000000..3ab4bd2a1
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/keystate/state/proxy/ValueStateStoreManagerProxy.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.keystate.state.proxy;
+
+import io.ray.streaming.state.KeyValueState;
+import io.ray.streaming.state.backend.KeyStateBackend;
+import io.ray.streaming.state.keystate.desc.ValueStateDescriptor;
+import io.ray.streaming.state.keystate.state.ValueState;
+import io.ray.streaming.state.keystate.state.impl.ValueStateImpl;
+import io.ray.streaming.state.strategy.StateStoreManagerProxy;
+
+/**
+ * This class defines ValueState Wrapper, connecting state and backend.
+ */
+public class ValueStateStoreManagerProxy extends StateStoreManagerProxy implements
+ KeyValueState {
+
+ private final ValueStateImpl valueState;
+
+ public ValueStateStoreManagerProxy(KeyStateBackend keyStateBackend,
+ ValueStateDescriptor stateDescriptor) {
+ super(keyStateBackend, stateDescriptor);
+ this.valueState = new ValueStateImpl<>(stateDescriptor, keyStateBackend);
+ }
+
+ public ValueState getValueState() {
+ return this.valueState;
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/serialization/KeyMapStoreSerializer.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/serialization/KeyMapStoreSerializer.java
new file mode 100644
index 000000000..620913a4c
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/serialization/KeyMapStoreSerializer.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.serialization;
+
+/**
+ * Key Map Serialization and Deserialization.
+ */
+public interface KeyMapStoreSerializer {
+
+ byte[] serializeKey(K key);
+
+ byte[] serializeUKey(S uk);
+
+ S deserializeUKey(byte[] ukArray);
+
+ byte[] serializeUValue(T uv);
+
+ T deserializeUValue(byte[] uvArray);
+
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/serialization/KeyValueStoreSerialization.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/serialization/KeyValueStoreSerialization.java
new file mode 100644
index 000000000..dc53ef7ba
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/serialization/KeyValueStoreSerialization.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.serialization;
+
+/**
+ * Key Value Serialization and Deserialization.
+ */
+public interface KeyValueStoreSerialization {
+
+ byte[] serializeKey(K key);
+
+ byte[] serializeValue(V value);
+
+ V deserializeValue(byte[] valueArray);
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/serialization/Serializer.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/serialization/Serializer.java
new file mode 100644
index 000000000..6db0c589f
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/serialization/Serializer.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.serialization;
+
+import org.nustaq.serialization.FSTConfiguration;
+
+/**
+ * fst wrapper.
+ */
+public class Serializer {
+
+ private static final ThreadLocal conf = ThreadLocal
+ .withInitial(FSTConfiguration::createDefaultConfiguration);
+
+ public static byte[] object2Bytes(Object value) {
+ return conf.get().asByteArray(value);
+ }
+
+ public static Object bytes2Object(byte[] buffer) {
+ return conf.get().asObject(buffer);
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/serialization/impl/AbstractSerialization.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/serialization/impl/AbstractSerialization.java
new file mode 100644
index 000000000..cc8daee6e
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/serialization/impl/AbstractSerialization.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.serialization.impl;
+
+import com.google.common.hash.Hashing;
+import io.ray.streaming.state.StateException;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * AbstractSerialization. Generate row key.
+ */
+public abstract class AbstractSerialization {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractSerialization.class);
+
+ public String generateRowKeyPrefix(String key) {
+ if (StringUtils.isNotEmpty(key)) {
+ String md5 = Hashing.md5().hashUnencodedChars(key).toString();
+ if ("".equals(md5)) {
+ throw new StateException("Invalid value to md5:" + key);
+ }
+ return StringUtils.substring(md5, 0, 4) + ":" + key;
+ } else {
+ LOG.warn("key is empty");
+ return key;
+ }
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/serialization/impl/DefaultKeyMapStoreSerializer.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/serialization/impl/DefaultKeyMapStoreSerializer.java
new file mode 100644
index 000000000..99f51764b
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/serialization/impl/DefaultKeyMapStoreSerializer.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.serialization.impl;
+
+import io.ray.streaming.state.serialization.KeyMapStoreSerializer;
+import io.ray.streaming.state.serialization.Serializer;
+
+/**
+ * Default Key Map Serialization and Deserialization.
+ */
+public class DefaultKeyMapStoreSerializer extends AbstractSerialization implements
+ KeyMapStoreSerializer {
+
+ @Override
+ public byte[] serializeKey(K key) {
+ String keyWithPrefix = generateRowKeyPrefix(key.toString());
+ return keyWithPrefix.getBytes();
+ }
+
+ @Override
+ public byte[] serializeUKey(S uk) {
+ return Serializer.object2Bytes(uk);
+ }
+
+ @Override
+ public S deserializeUKey(byte[] ukArray) {
+ return (S) Serializer.bytes2Object(ukArray);
+ }
+
+ @Override
+ public byte[] serializeUValue(T uv) {
+ return Serializer.object2Bytes(uv);
+ }
+
+ @Override
+ public T deserializeUValue(byte[] uvArray) {
+ return (T) Serializer.bytes2Object(uvArray);
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/serialization/impl/DefaultKeyValueStoreSerialization.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/serialization/impl/DefaultKeyValueStoreSerialization.java
new file mode 100644
index 000000000..ff0a843f6
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/serialization/impl/DefaultKeyValueStoreSerialization.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.serialization.impl;
+
+import io.ray.streaming.state.serialization.KeyValueStoreSerialization;
+import io.ray.streaming.state.serialization.Serializer;
+
+/**
+ * KV Store Serialization and Deserialization.
+ */
+public class DefaultKeyValueStoreSerialization extends AbstractSerialization implements
+ KeyValueStoreSerialization {
+
+ @Override
+ public byte[] serializeKey(K key) {
+ String keyWithPrefix = generateRowKeyPrefix(key.toString());
+ return keyWithPrefix.getBytes();
+ }
+
+ @Override
+ public byte[] serializeValue(V value) {
+ return Serializer.object2Bytes(value);
+ }
+
+ @Override
+ public V deserializeValue(byte[] valueArray) {
+ return (V) Serializer.bytes2Object(valueArray);
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/store/KeyMapStore.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/store/KeyMapStore.java
new file mode 100644
index 000000000..0393ee294
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/store/KeyMapStore.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.store;
+
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * Key Map Store interface.
+ */
+public interface KeyMapStore extends KeyValueStore> {
+
+ /**
+ * put sub key value into the store incrementally.
+ */
+ void put(K key, S subKey, T value) throws IOException;
+
+ /**
+ * get subValue from store.
+ */
+ T get(K key, S subKey) throws IOException;
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/store/KeyValueStore.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/store/KeyValueStore.java
new file mode 100644
index 000000000..506baef34
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/store/KeyValueStore.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.store;
+
+import java.io.IOException;
+import java.io.Serializable;
+
+/**
+ * Key Value Store interface.
+ */
+public interface KeyValueStore extends Serializable {
+
+ /**
+ * put key value into store.
+ */
+ void put(K key, V value) throws IOException;
+
+ /**
+ * get value from store.
+ */
+ V get(K key) throws IOException;
+
+ /**
+ * remove key in the store.
+ */
+ void remove(K key) throws IOException;
+
+ /**
+ * flush to disk.
+ */
+ void flush() throws IOException;
+
+ /**
+ * clear all cache.
+ */
+ void clearCache();
+
+ /**
+ * close the store.
+ */
+ void close() throws IOException;
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/store/impl/MemoryKeyMapStore.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/store/impl/MemoryKeyMapStore.java
new file mode 100644
index 000000000..aad605c5a
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/store/impl/MemoryKeyMapStore.java
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.store.impl;
+
+import com.google.common.collect.Maps;
+import io.ray.streaming.state.store.KeyMapStore;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Memory Key Map Store.
+ */
+public class MemoryKeyMapStore implements KeyMapStore {
+
+ private Map> memoryStore;
+
+ public MemoryKeyMapStore() {
+ this.memoryStore = Maps.newConcurrentMap();
+ }
+
+ @Override
+ public void put(K key, Map value) throws IOException {
+ this.memoryStore.put(key, value);
+ }
+
+ @Override
+ public void put(K key, S subKey, T value) throws IOException {
+ if (memoryStore.containsKey(key)) {
+ memoryStore.get(key).put(subKey, value);
+ } else {
+ Map map = new HashMap<>();
+ map.put(subKey, value);
+ memoryStore.put(key, map);
+ }
+ }
+
+ @Override
+ public Map get(K key) throws IOException {
+ return this.memoryStore.get(key);
+ }
+
+ @Override
+ public T get(K key, S subKey) throws IOException {
+ if (memoryStore.containsKey(key)) {
+ return memoryStore.get(key).get(subKey);
+ }
+ return null;
+ }
+
+ @Override
+ public void remove(K key) throws IOException {
+ this.memoryStore.remove(key);
+ }
+
+ @Override
+ public void flush() throws IOException {
+
+ }
+
+ @Override
+ public void clearCache() {
+
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (memoryStore != null) {
+ memoryStore.clear();
+ }
+ }
+
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/store/impl/MemoryKeyValueStore.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/store/impl/MemoryKeyValueStore.java
new file mode 100644
index 000000000..d5882832c
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/store/impl/MemoryKeyValueStore.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.store.impl;
+
+import com.google.common.collect.Maps;
+import io.ray.streaming.state.store.KeyValueStore;
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * Memory Key Value Store.
+ */
+public class MemoryKeyValueStore implements KeyValueStore {
+
+ private Map memoryStore;
+
+ public MemoryKeyValueStore() {
+ this.memoryStore = Maps.newConcurrentMap();
+ }
+
+ @Override
+ public void put(K key, V value) throws IOException {
+ this.memoryStore.put(key, value);
+ }
+
+ @Override
+ public V get(K key) throws IOException {
+ return this.memoryStore.get(key);
+ }
+
+ @Override
+ public void remove(K key) throws IOException {
+ this.memoryStore.remove(key);
+ }
+
+ @Override
+ public void flush() throws IOException {
+
+ }
+
+ @Override
+ public void clearCache() {
+
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (memoryStore != null) {
+ memoryStore.clear();
+ }
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/strategy/AbstractStateStoreManager.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/strategy/AbstractStateStoreManager.java
new file mode 100644
index 000000000..0d4e179fe
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/strategy/AbstractStateStoreManager.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.strategy;
+
+import io.ray.streaming.state.StateException;
+import io.ray.streaming.state.StateStoreManager;
+import io.ray.streaming.state.StorageRecord;
+import io.ray.streaming.state.serialization.Serializer;
+import io.ray.streaming.state.store.KeyValueStore;
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * This class defines the StoreManager Abstract class.
+ * We use three layer to store the state, frontStore, middleStore and keyValueStore(remote).
+ */
+public abstract class AbstractStateStoreManager implements StateStoreManager {
+
+ /**
+ * read-write
+ */
+ protected Map> frontStore = new ConcurrentHashMap<>();
+
+ /**
+ * remote-storage
+ */
+ protected KeyValueStore> kvStore;
+
+ /**
+ * read-only
+ */
+ protected Map> middleStore = new ConcurrentHashMap<>();
+ protected int keyGroupIndex = -1;
+
+ public AbstractStateStoreManager(KeyValueStore> backStore) {
+ kvStore = backStore;
+ }
+
+ public byte[] toBytes(StorageRecord storageRecord) {
+ return Serializer.object2Bytes(storageRecord);
+ }
+
+ public StorageRecord toStorageRecord(byte[] data) {
+ return (StorageRecord) Serializer.bytes2Object(data);
+ }
+
+ public abstract V get(long checkpointId, String key);
+
+ public void put(long checkpointId, String k, V v) {
+ frontStore.put(k, new StorageRecord<>(checkpointId, v));
+ }
+
+ @Override
+ public void ackCommit(long checkpointId, long timeStamp) {
+ ackCommit(checkpointId);
+ }
+
+ public abstract void ackCommit(long checkpointId);
+
+ public void setKeyGroupIndex(int keyGroupIndex) {
+ this.keyGroupIndex = keyGroupIndex;
+ }
+
+ public void close() {
+ frontStore.clear();
+ middleStore.clear();
+ if (kvStore != null) {
+ kvStore.clearCache();
+ try {
+ kvStore.close();
+ } catch (IOException e) {
+ throw new StateException(e);
+ }
+ }
+ }
+}
diff --git a/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/strategy/DualStateStoreManager.java b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/strategy/DualStateStoreManager.java
new file mode 100644
index 000000000..f3f31b1b2
--- /dev/null
+++ b/streaming/java/streaming-state/src/main/java/io/ray/streaming/state/strategy/DualStateStoreManager.java
@@ -0,0 +1,167 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.ray.streaming.state.strategy;
+
+import com.google.common.primitives.Longs;
+import io.ray.streaming.state.StateException;
+import io.ray.streaming.state.StorageRecord;
+import io.ray.streaming.state.store.KeyValueStore;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class define the checkpoint store strategy, which saves two-version data once.
+ */
+public class DualStateStoreManager