wombatu-kun commented on code in PR #17013: URL: https://github.com/apache/iceberg/pull/17013#discussion_r3503950986
########## core/src/test/java/org/apache/iceberg/hadoop/TestHadoopTableOperations.java: ########## @@ -0,0 +1,385 @@ +/* + * 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.iceberg.hadoop; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.io.ByteArrayInputStream; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.Map; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.fs.FSInputStream; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.PathFilter; +import org.apache.hadoop.fs.permission.FsPermission; +import org.apache.hadoop.util.Progressable; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.junit.jupiter.api.Test; + +public class TestHadoopTableOperations { + @Test + public void testVersionHintRetryBeforeMetadataScan() { + Configuration conf = new Configuration(); + conf.setInt(ConfigProperties.VERSION_HINT_NUM_RETRIES, 2); + conf.setLong(ConfigProperties.VERSION_HINT_RETRY_MIN_WAIT_MS, 0L); + + TestingFileSystem fs = new TestingFileSystem(); + fs.addDirectory("metadata"); + fs.addFile("v3.metadata.json", ""); + fs.addVersionHintAvailableAfterAttempt(3, "7"); + + HadoopTableOperations ops = newTableOps(conf, fs); + + assertThat(ops.findVersion()).isEqualTo(7); + assertThat(fs.versionHintOpenAttempts()).isEqualTo(3); + assertThat(fs.metadataRootExistsCalls()).isEqualTo(1); + assertThat(fs.listStatusCalls()).isZero(); + } + + @Test + public void testVersionHintRetriesCanBeDisabled() { + Configuration conf = new Configuration(); + conf.setInt(ConfigProperties.VERSION_HINT_NUM_RETRIES, 0); + + TestingFileSystem fs = new TestingFileSystem(); + fs.addDirectory("metadata"); + fs.addFile("v1.metadata.json", ""); + fs.addFile("v3.metadata.json", ""); + + HadoopTableOperations ops = newTableOps(conf, fs); + + assertThat(ops.findVersion()).isEqualTo(3); + assertThat(fs.versionHintOpenAttempts()).isEqualTo(1); + assertThat(fs.metadataRootExistsCalls()).isEqualTo(1); + assertThat(fs.listStatusCalls()).isEqualTo(1); + } + + @Test + public void testConfirmedMetadataRootIsNotCheckedAgainBeforeFallbackScan() { + Configuration conf = new Configuration(); + conf.setInt(ConfigProperties.VERSION_HINT_NUM_RETRIES, 2); + conf.setLong(ConfigProperties.VERSION_HINT_RETRY_MIN_WAIT_MS, 0L); + + TestingFileSystem fs = new TestingFileSystem(); + fs.addDirectory("metadata"); + fs.addFile("v3.metadata.json", ""); + fs.failMetadataRootExistsAfterCall(1); Review Comment: No test covers the `metadataRootCheckFailure` branch - when `fs.exists(metadataRoot())` throws on its first probe during the retry phase. This test uses `failMetadataRootExistsAfterCall(1)`, so probe #1 succeeds and caches TRUE and the throwing case is never hit. Add a sibling with `failMetadataRootExistsAfterCall(0)` asserting `findVersion() == 0`, `metadataRootExistsCalls() == 1`, and `listStatusCalls() == 0`, to pin the "do not re-probe a failing filesystem" contract. ########## core/src/main/java/org/apache/iceberg/hadoop/HadoopTableOperations.java: ########## @@ -313,43 +315,172 @@ private void writeVersionToPath(FileSystem fs, Path path, int versionToWrite) th int findVersion() { Path versionHintFile = versionHintFile(); FileSystem fs = getFileSystem(versionHintFile, conf); + VersionHintRetryState retryState = new VersionHintRetryState(); - try (InputStreamReader fsr = - new InputStreamReader(fs.open(versionHintFile), StandardCharsets.UTF_8); - BufferedReader in = new BufferedReader(fsr)) { - return Integer.parseInt(in.readLine().replace("\n", "")); - - } catch (Exception e) { - try { - if (fs.exists(metadataRoot())) { - LOG.warn("Error reading version hint file {}", versionHintFile, e); - } else { - LOG.debug("Metadata for table not found in directory {}", metadataRoot(), e); - return 0; - } + try { + return retryReadVersionHint(fs, versionHintFile, retryState); - // List the metadata directory to find the version files, and try to recover the max - // available version - FileStatus[] files = - fs.listStatus( - metadataRoot(), name -> VERSION_PATTERN.matcher(name.getName()).matches()); - int maxVersion = 0; - - for (FileStatus file : files) { - int currentVersion = version(file.getPath().getName()); - if (currentVersion > maxVersion && getMetadataFile(currentVersion) != null) { - maxVersion = currentVersion; - } - } + } catch (RuntimeException e) { + if (e.getCause() instanceof InterruptedException) { + throw e; + } - return maxVersion; - } catch (IOException io) { - LOG.warn("Error trying to recover the latest version number for {}", versionHintFile, io); + return findVersionFromMetadataDirectory(fs, versionHintFile, e, retryState); + } catch (IOException e) { + return findVersionFromMetadataDirectory(fs, versionHintFile, e, retryState); + } + } + + private int findVersionFromMetadataDirectory( + FileSystem fs, + Path versionHintFile, + Exception versionHintFailure, + VersionHintRetryState retryState) { + if (retryState.metadataRootCheckFailure != null) { Review Comment: The metadata-root fallback is over-structured. `metadataRootCheckFailure` and this first `findVersionFromMetadataDirectory` overload exist only so an `fs.exists()` failure during retry warns and returns 0; but if `shouldRetryReadVersionHint` returned false on that IOException without recording it, the fallback would reach the boolean overload, re-probe `metadataRootExists`, hit the same throw, and the existing `catch (IOException)` would produce the identical WARN + `return 0`. Dropping the field collapses the two same-named overloads (they differ only by the 4th arg type) into one and removes the duplicated "Error trying to recover the latest version number" log; the only cost is one extra `fs.exists()` on the already-failing path. Separately, `metadataRootExists` is at once a field, a method, and a boolean parameter (they collide on line 362) - renaming the check method (e.g. `metadataRootPresent`) removes that ambiguity. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
