[
https://issues.apache.org/jira/browse/FLINK-10918?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16697867#comment-16697867
]
ASF GitHub Bot commented on FLINK-10918:
----------------------------------------
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:
us...@infra.apache.org
> incremental Keyed state with RocksDB throws cannot create directory error in
> windows
> ------------------------------------------------------------------------------------
>
> Key: FLINK-10918
> URL: https://issues.apache.org/jira/browse/FLINK-10918
> Project: Flink
> Issue Type: Bug
> Components: Queryable State, State Backends, Checkpointing
> Affects Versions: 1.6.2
> Environment: windows
> {code:java}
> val env = StreamExecutionEnvironment.getExecutionEnvironment
> val rocksdb = new RocksDBStateBackend("file:\\C:\\rocksdb\\checkpoint",true)
> rocksdb.setDbStoragePath("file:\\C:\\rocksdb\\storage")
> env.setStateBackend(rocksdb)
> env.enableCheckpointing(10)
> {code}
>
> Reporter: Amit
> Priority: Major
> Labels: pull-request-available
> Attachments:
> 0001-FLINK-10918-Fix-for-checkpoint-creation-on-windows-1.patch
>
>
> Facing error while enabling keyed state with RocksDBBackend with
> checkpointing to a local windows directory
>
> {code:java}
> Caused by: org.rocksdb.RocksDBException: Failed to create dir:
> /c:/tmp/data/job_dbe01128760d4d5cb90809cd94c2a936_op_StreamMap_b5c8d46f3e7b141acf271f12622e752b__3_8__uuid_45c1f62b-a198-44f5-add5-7683079b03f8/chk-1.tmp:
> Invalid argument
> at org.rocksdb.Checkpoint.createCheckpoint(Native Method)
> at org.rocksdb.Checkpoint.createCheckpoint(Checkpoint.java:51)
> at
> org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend$RocksDBIncrementalSnapshotOperation.takeSnapshot(RocksDBKeyedStateBackend.java:2549)
> at
> org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend$IncrementalSnapshotStrategy.performSnapshot(RocksDBKeyedStateBackend.java:2008)
> at
> org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend.snapshot(RocksDBKeyedStateBackend.java:498)
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:406)
> ... 13 more
> {code}
>
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
