flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gyf...@apache.org
Subject flink git commit: [FLINK-3798] [streaming] Clean up RocksDB backend field/method access
Date Thu, 21 Apr 2016 09:33:19 GMT
Repository: flink
Updated Branches:
  refs/heads/master a43bade0d -> d636bf78e


[FLINK-3798] [streaming] Clean up RocksDB backend field/method access

Closes #1918


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/d636bf78
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/d636bf78
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/d636bf78

Branch: refs/heads/master
Commit: d636bf78edffc490586b99f259beac1bb039f580
Parents: a43bade
Author: Gyula Fora <gyfora@apache.org>
Authored: Thu Apr 21 09:20:57 2016 +0200
Committer: Gyula Fora <gyfora@apache.org>
Committed: Thu Apr 21 11:30:03 2016 +0200

----------------------------------------------------------------------
 .../contrib/streaming/state/AbstractRocksDBState.java   | 12 ++++++------
 .../contrib/streaming/state/RocksDBFoldingState.java    |  2 +-
 .../flink/contrib/streaming/state/RocksDBListState.java |  2 +-
 .../contrib/streaming/state/RocksDBReducingState.java   |  2 +-
 .../contrib/streaming/state/RocksDBStateBackend.java    |  8 ++++----
 .../contrib/streaming/state/RocksDBValueState.java      |  2 +-
 6 files changed, 14 insertions(+), 14 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/d636bf78/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/AbstractRocksDBState.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/AbstractRocksDBState.java
b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/AbstractRocksDBState.java
index ec58a63..5d7e1d8 100644
--- a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/AbstractRocksDBState.java
+++ b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/AbstractRocksDBState.java
@@ -60,14 +60,14 @@ abstract class AbstractRocksDBState<K, N, S extends State, SD extends
StateDescr
 	protected RocksDBStateBackend backend;
 
 	/** The column family of this particular instance of state */
-	ColumnFamilyHandle columnFamily;
+	protected ColumnFamilyHandle columnFamily;
 
 	/**
 	 * Creates a new RocksDB backed state.
 	 *
 	 * @param namespaceSerializer The serializer for the namespace.
 	 */
-	AbstractRocksDBState(ColumnFamilyHandle columnFamily,
+	protected AbstractRocksDBState(ColumnFamilyHandle columnFamily,
 			TypeSerializer<N> namespaceSerializer,
 			RocksDBStateBackend backend) {
 
@@ -80,7 +80,7 @@ abstract class AbstractRocksDBState<K, N, S extends State, SD extends
StateDescr
 	// ------------------------------------------------------------------------
 
 	@Override
-	final public void clear() {
+	public void clear() {
 		ByteArrayOutputStream baos = new ByteArrayOutputStream();
 		DataOutputViewStreamWrapper out = new DataOutputViewStreamWrapper(baos);
 		try {
@@ -92,19 +92,19 @@ abstract class AbstractRocksDBState<K, N, S extends State, SD extends
StateDescr
 		}
 	}
 
-	void writeKeyAndNamespace(DataOutputView out) throws IOException {
+	protected void writeKeyAndNamespace(DataOutputView out) throws IOException {
 		backend.keySerializer().serialize(backend.currentKey(), out);
 		out.writeByte(42);
 		namespaceSerializer.serialize(currentNamespace, out);
 	}
 
 	@Override
-	final public void setCurrentNamespace(N namespace) {
+	public void setCurrentNamespace(N namespace) {
 		this.currentNamespace = namespace;
 	}
 
 	@Override
-	final public void dispose() {
+	public void dispose() {
 		// ignore because we don't hold any state ourselves
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/d636bf78/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBFoldingState.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBFoldingState.java
b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBFoldingState.java
index 91fa807..411077f 100644
--- a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBFoldingState.java
+++ b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBFoldingState.java
@@ -69,7 +69,7 @@ class RocksDBFoldingState<K, N, T, ACC>
 	 * @param stateDesc The state identifier for the state. This contains name
 	 *                     and can create a default state value.
 	 */
-	RocksDBFoldingState(ColumnFamilyHandle columnFamily,
+	public RocksDBFoldingState(ColumnFamilyHandle columnFamily,
 			TypeSerializer<N> namespaceSerializer,
 			FoldingStateDescriptor<T, ACC> stateDesc,
 			RocksDBStateBackend backend) {

http://git-wip-us.apache.org/repos/asf/flink/blob/d636bf78/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java
b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java
index dc55d11..77c3c3e 100644
--- a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java
+++ b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBListState.java
@@ -71,7 +71,7 @@ class RocksDBListState<K, N, V>
 	 * @param stateDesc The state identifier for the state. This contains name
 	 *                     and can create a default state value.
 	 */
-	RocksDBListState(ColumnFamilyHandle columnFamily,
+	public RocksDBListState(ColumnFamilyHandle columnFamily,
 			TypeSerializer<N> namespaceSerializer,
 			ListStateDescriptor<V> stateDesc,
 			RocksDBStateBackend backend) {

http://git-wip-us.apache.org/repos/asf/flink/blob/d636bf78/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBReducingState.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBReducingState.java
b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBReducingState.java
index 953660d..7a4655a 100644
--- a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBReducingState.java
+++ b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBReducingState.java
@@ -68,7 +68,7 @@ class RocksDBReducingState<K, N, V>
 	 * @param stateDesc The state identifier for the state. This contains name
 	 *                     and can create a default state value.
 	 */
-	RocksDBReducingState(ColumnFamilyHandle columnFamily,
+	public RocksDBReducingState(ColumnFamilyHandle columnFamily,
 			TypeSerializer<N> namespaceSerializer,
 			ReducingStateDescriptor<V> stateDesc,
 			RocksDBStateBackend backend) {

http://git-wip-us.apache.org/repos/asf/flink/blob/d636bf78/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
index 3d63bd2..99ee1cd 100644
--- a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
+++ b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBStateBackend.java
@@ -155,7 +155,7 @@ public class RocksDBStateBackend extends AbstractStateBackend {
 	 * to store state. The different k/v states that we have don't each have their own RocksDB
 	 * instance. They all write to this instance but to their own column family.
 	 */
-	transient RocksDB db;
+	protected transient RocksDB db;
 
 	/**
 	 * Information about the k/v states as we create them. This is used to retrieve the
@@ -807,7 +807,7 @@ public class RocksDBStateBackend extends AbstractStateBackend {
 	 * <p>This also checks whether the {@link StateDescriptor} for a state matches the
one
 	 * that we checkpointed, i.e. is already in the map of column families.
 	 */
-	private ColumnFamilyHandle getColumnFamily(StateDescriptor descriptor)  {
+	protected ColumnFamilyHandle getColumnFamily(StateDescriptor descriptor)  {
 
 		Tuple2<ColumnFamilyHandle, StateDescriptor> stateInfo = kvStateInformation.get(descriptor.getName());
 
@@ -832,14 +832,14 @@ public class RocksDBStateBackend extends AbstractStateBackend {
 	/**
 	 * Used by k/v states to access the current key.
 	 */
-	Object currentKey() {
+	public Object currentKey() {
 		return currentKey;
 	}
 
 	/**
 	 * Used by k/v states to access the key serializer.
 	 */
-	TypeSerializer keySerializer() {
+	public TypeSerializer keySerializer() {
 		return keySerializer;
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/d636bf78/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBValueState.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBValueState.java
b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBValueState.java
index 2af06ad..47d86b8 100644
--- a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBValueState.java
+++ b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBValueState.java
@@ -64,7 +64,7 @@ class RocksDBValueState<K, N, V>
 	 * @param stateDesc The state identifier for the state. This contains name
 	 *                           and can create a default state value.
 	 */
-	RocksDBValueState(ColumnFamilyHandle columnFamily,
+	public RocksDBValueState(ColumnFamilyHandle columnFamily,
 			TypeSerializer<N> namespaceSerializer,
 			ValueStateDescriptor<V> stateDesc,
 			RocksDBStateBackend backend) {


Mime
View raw message