This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new 84f14ee9 [hotfix] Fix AtomicFileWriter useage bug
84f14ee9 is described below
commit 84f14ee9527d59138b476fa22188d248b6ffb11f
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Jul 5 10:32:18 2022 +0800
[hotfix] Fix AtomicFileWriter useage bug
This closes #190
---
.../store/file/operation/FileStoreCommitImpl.java | 4 +-
.../table/store/file/schema/SchemaManager.java | 4 +-
.../table/store/file/utils/AtomicFileWriter.java | 23 +++++++++
.../table/store/file/utils/MetaFileWriter.java | 58 ----------------------
...leWriterTest.java => AtomicFileWriterTest.java} | 37 +++-----------
5 files changed, 33 insertions(+), 93 deletions(-)
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreCommitImpl.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreCommitImpl.java
index 58b5739f..7e9b8df7 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreCommitImpl.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreCommitImpl.java
@@ -32,8 +32,8 @@ import
org.apache.flink.table.store.file.manifest.ManifestFileMeta;
import org.apache.flink.table.store.file.manifest.ManifestList;
import org.apache.flink.table.store.file.predicate.Predicate;
import org.apache.flink.table.store.file.predicate.PredicateConverter;
+import org.apache.flink.table.store.file.utils.AtomicFileWriter;
import org.apache.flink.table.store.file.utils.FileStorePathFactory;
-import org.apache.flink.table.store.file.utils.MetaFileWriter;
import org.apache.flink.table.store.file.utils.SnapshotManager;
import org.apache.flink.table.store.utils.RowDataToObjectArrayConverter;
import org.apache.flink.table.types.logical.RowType;
@@ -401,7 +401,7 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
}
boolean committed =
- MetaFileWriter.writeFileSafety(
+ AtomicFileWriter.writeFileUtf8(
newSnapshotPath, newSnapshot.toJson());
if (committed) {
snapshotManager.commitLatestHint(newSnapshotId);
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/SchemaManager.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/SchemaManager.java
index 7132d9dc..ad4a13c5 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/SchemaManager.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/SchemaManager.java
@@ -21,9 +21,9 @@ package org.apache.flink.table.store.file.schema;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
import org.apache.flink.table.store.file.operation.Lock;
+import org.apache.flink.table.store.file.utils.AtomicFileWriter;
import org.apache.flink.table.store.file.utils.FileUtils;
import org.apache.flink.table.store.file.utils.JsonSerdeUtil;
-import org.apache.flink.table.store.file.utils.MetaFileWriter;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.util.Preconditions;
@@ -139,7 +139,7 @@ public class SchemaManager implements Serializable {
return false;
}
- return MetaFileWriter.writeFileSafety(
+ return AtomicFileWriter.writeFileUtf8(
schemaPath, tableSchema.toString());
});
if (success) {
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/AtomicFileWriter.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/AtomicFileWriter.java
index 1363ec3a..4e3467fd 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/AtomicFileWriter.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/AtomicFileWriter.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.store.file.utils;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.FileSystemKind;
import org.apache.flink.core.fs.Path;
@@ -54,4 +55,26 @@ public interface AtomicFileWriter {
return new RecoverableAtomicFileWriter(recoverableWriter);
}
}
+
+ /**
+ * Write an utf8 string to file.
+ *
+ * @return True if the committing was successful, False otherwise
+ */
+ static boolean writeFileUtf8(Path path, String content) throws IOException
{
+ return writeFileUtf8(AtomicFileWriter.create(path.getFileSystem()),
path, content);
+ }
+
+ @VisibleForTesting
+ static boolean writeFileUtf8(AtomicFileWriter writer, Path path, String
content)
+ throws IOException {
+ AtomicFsDataOutputStream out = writer.open(path);
+ try {
+ FileUtils.writeOutputStreamUtf8(out, content);
+ return out.closeAndCommit();
+ } catch (IOException e) {
+ out.close();
+ throw e;
+ }
+ }
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/MetaFileWriter.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/MetaFileWriter.java
deleted file mode 100644
index b88f2a30..00000000
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/MetaFileWriter.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * 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 org.apache.flink.table.store.file.utils;
-
-import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.core.fs.Path;
-
-import java.io.IOException;
-
-/** File writer to write a metadata file which contains only a small amount of
text data. */
-public class MetaFileWriter {
-
- /**
- * Write an utf8 string to file. In addition to that, it checks if the
committed file is equal
- * to the input {@code content}, if not, the committing has failed.
- *
- * <p>But this does not solve overwritten committing.
- *
- * @return True if the committing was successful, False otherwise
- */
- public static boolean writeFileSafety(Path path, String content) throws
IOException {
- return writeFileSafety(AtomicFileWriter.create(path.getFileSystem()),
path, content);
- }
-
- @VisibleForTesting
- static boolean writeFileSafety(AtomicFileWriter writer, Path path, String
content)
- throws IOException {
- AtomicFsDataOutputStream out = writer.open(path);
- try {
- FileUtils.writeOutputStreamUtf8(out, content);
- boolean success = out.closeAndCommit();
- if (success) {
- return content.equals(FileUtils.readFileUtf8(path));
- } else {
- return false;
- }
- } catch (IOException e) {
- out.close();
- return false;
- }
- }
-}
diff --git
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/MetaFileWriterTest.java
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/AtomicFileWriterTest.java
similarity index 56%
rename from
flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/MetaFileWriterTest.java
rename to
flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/AtomicFileWriterTest.java
index 476e9c77..8e94febb 100644
---
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/MetaFileWriterTest.java
+++
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/AtomicFileWriterTest.java
@@ -25,14 +25,13 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
-import java.util.UUID;
+import static
org.apache.flink.table.store.file.utils.AtomicFileWriter.writeFileUtf8;
import static org.apache.flink.table.store.file.utils.FileUtils.readFileUtf8;
-import static
org.apache.flink.table.store.file.utils.MetaFileWriter.writeFileSafety;
import static org.assertj.core.api.Assertions.assertThat;
-/** Test for {@link MetaFileWriter}. */
-public class MetaFileWriterTest {
+/** Test for {@link AtomicFileWriter}. */
+public class AtomicFileWriterTest {
@TempDir java.nio.file.Path tempDir;
@@ -45,37 +44,13 @@ public class MetaFileWriterTest {
@Test
public void testDefault() throws IOException {
- test(root, AtomicFileWriter.create(root.getFileSystem()));
- }
-
- @Test
- public void testSafety() throws IOException {
- AtomicFileWriter writer =
- path ->
- new RenamingAtomicFsDataOutputStream(
- path.getFileSystem(),
- path,
- new Path(
- path.getParent(),
- "." + path.getName() +
UUID.randomUUID())) {
- @Override
- public boolean closeAndCommit() throws IOException
{
- super.closeAndCommit();
-
- // always return true
- return true;
- }
- };
- test(root, writer);
- }
-
- private void test(Path root, AtomicFileWriter writer) throws IOException {
+ AtomicFileWriter writer =
AtomicFileWriter.create(root.getFileSystem());
Path path1 = new Path(root, "f1");
- boolean success = writeFileSafety(writer, path1, "hahaha");
+ boolean success = writeFileUtf8(writer, path1, "hahaha");
assertThat(success).isTrue();
assertThat(readFileUtf8(path1)).isEqualTo("hahaha");
- success = writeFileSafety(writer, path1, "xixixi");
+ success = writeFileUtf8(writer, path1, "xixixi");
assertThat(success).isFalse();
}
}
