[GitHub] imamitsehgal closed pull request #7153: FLINK-10918 Fix the checkpoint dir creation error on Window for keyed rocksdb state

[GitBox <gi...@apache.org>]

imamitsehgal closed pull request #7153: FLINK-10918 Fix the checkpoint dir creation error on Window for keyed rocksdb state
URL: https://github.com/apache/flink/pull/7153
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
index 17ba985f883..4db32e1d2c8 100644
--- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
+++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
@@ -557,6 +557,14 @@ private void createDB() throws IOException {
 		this.defaultColumnFamily = columnFamilyHandles.get(0);
 	}
 
+	private String getPathAsString(Path path) {
+		String pathString = path.getPath();
+		if (path.hasWindowsDrive() && pathString.startsWith("/")) {
+			pathString = pathString.substring(1);
+		}
+		return pathString;
+	}
+
 	private RocksDB openDB(
 		String path,
 		List<ColumnFamilyDescriptor> stateColumnFamilyDescriptors,
@@ -575,7 +583,7 @@ private RocksDB openDB(
 		try {
 			dbRef = RocksDB.open(
 				Preconditions.checkNotNull(dbOptions),
-				Preconditions.checkNotNull(path),
+				Preconditions.checkNotNull(getPathAsString(new Path(path))),
 				columnFamilyDescriptors,
 				stateColumnFamilyHandles);
 		} catch (RocksDBException e) {
@@ -2374,6 +2382,7 @@ private void closeLocalRegistry() {
 		}
 	}
 
+
 	/**
 	 * Encapsulates the process to perform an incremental snapshot of a RocksDBKeyedStateBackend.
 	 */
@@ -2546,7 +2555,9 @@ void takeSnapshot() throws Exception {
 
 			// create hard links of living files in the snapshot path
 			try (Checkpoint checkpoint = Checkpoint.create(stateBackend.db)) {
-				checkpoint.createCheckpoint(localBackupDirectory.getDirectory().getPath());
+				String path = stateBackend.getPathAsString(localBackupDirectory.getDirectory());
+				LOG.trace("Checkpoint path is {}.", path);
+				checkpoint.createCheckpoint(path);
 			}
 		}
 


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services