diff --git a/common/src/main/java/org/apache/rocketmq/common/utils/CheckpointFile.java b/common/src/main/java/org/apache/rocketmq/common/utils/CheckpointFile.java index 8f51d274654..b161cec62eb 100644 --- a/common/src/main/java/org/apache/rocketmq/common/utils/CheckpointFile.java +++ b/common/src/main/java/org/apache/rocketmq/common/utils/CheckpointFile.java @@ -103,11 +103,14 @@ private List read(String filePath) throws IOException { return result; } try (BufferedReader reader = Files.newBufferedReader(file.toPath())) { - // Read size - int expectedLines = Integer.parseInt(reader.readLine()); - - // Read block crc - int expectedCrc32 = Integer.parseInt(reader.readLine()); + int expectedLines; + int expectedCrc32; + try { + expectedLines = Integer.parseInt(reader.readLine()); + expectedCrc32 = Integer.parseInt(reader.readLine()); + } catch (NumberFormatException e) { + throw new IOException("Invalid checkpoint header, file=" + filePath, e); + } // Read entries StringBuilder sb = new StringBuilder(); diff --git a/common/src/test/java/org/apache/rocketmq/common/utils/CheckpointFileHeaderTest.java b/common/src/test/java/org/apache/rocketmq/common/utils/CheckpointFileHeaderTest.java new file mode 100644 index 00000000000..bd7cdf817e8 --- /dev/null +++ b/common/src/test/java/org/apache/rocketmq/common/utils/CheckpointFileHeaderTest.java @@ -0,0 +1,100 @@ +/* + * 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.rocketmq.common.utils; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.nio.file.StandardCopyOption; +import java.util.Collections; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +public class CheckpointFileHeaderTest { + + @Rule + public TemporaryFolder temporaryFolder = new TemporaryFolder(); + + private final CheckpointFile.CheckpointSerializer serializer = + new CheckpointFile.CheckpointSerializer() { + @Override + public String toLine(String entry) { + return entry; + } + + @Override + public String fromLine(String line) { + return line; + } + }; + + @Test + public void testTruncatedHeadersFallBackToValidBackup() throws IOException { + for (String header : new String[] {"", "1", "1\n", "1\n\nvalue\n"}) { + assertEquals(Collections.singletonList("saved"), withCorruptedPrimary(header).read()); + } + } + + @Test + public void testMalformedCountFallsBackToValidBackup() throws IOException { + assertEquals(Collections.singletonList("saved"), withCorruptedPrimary("invalid\n0\nvalue\n").read()); + } + + @Test + public void testMalformedCrcFallsBackToValidBackup() throws IOException { + assertEquals(Collections.singletonList("saved"), withCorruptedPrimary("1\ninvalid\nvalue\n").read()); + } + + @Test + public void testInvalidPrimaryAndBackupReportIOException() throws IOException { + CheckpointFile checkpoint = withCorruptedPrimary("invalid\n0\nvalue\n"); + Files.write(Paths.get(checkpoint.getBackFilePath()), "invalid".getBytes(StandardCharsets.UTF_8)); + IOException failure = assertThrows(IOException.class, checkpoint::read); + assertTrue(failure.getCause() instanceof NumberFormatException); + assertTrue(failure.getMessage().contains(checkpoint.getBackFilePath())); + } + + @Test + public void testCountAndCrcMismatchStillUseBackup() throws IOException { + for (String content : new String[] {"2\n0\nvalue\n", "1\n1\nvalue\n"}) { + assertEquals(Collections.singletonList("saved"), withCorruptedPrimary(content).read()); + } + } + + @Test + public void testValidPrimaryStillTakesPrecedence() throws IOException { + CheckpointFile checkpoint = withCorruptedPrimary("1\n0\ncurrent" + System.lineSeparator()); + assertEquals(Collections.singletonList("current"), checkpoint.read()); + } + + private CheckpointFile withCorruptedPrimary(String content) throws IOException { + Path path = temporaryFolder.newFile().toPath(); + CheckpointFile checkpoint = new CheckpointFile<>(path.toString(), serializer); + checkpoint.write(Collections.singletonList("saved")); + Files.copy(path, Paths.get(checkpoint.getBackFilePath()), StandardCopyOption.REPLACE_EXISTING); + Files.write(path, content.getBytes(StandardCharsets.UTF_8)); + return checkpoint; + } +}