Don't delete temp recovered checkpoint file if it was renamed

Closes #14872
This commit is contained in:
Boaz Leskes 2015-11-19 23:08:20 +01:00
parent b001e20a83
commit 6d9e82311a
1 changed files with 26 additions and 14 deletions

View File

@ -129,11 +129,11 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC
};
/**
* Creates a new Translog instance. This method will create a new transaction log unless the given {@link TranslogConfig} has
* a non-null {@link org.elasticsearch.index.translog.Translog.TranslogGeneration}. If the generation is null this method
* us destructive and will delete all files in the translog path given.
*
* @see TranslogConfig#getTranslogPath()
*/
public Translog(TranslogConfig config) throws IOException {
@ -141,7 +141,7 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC
this.config = config;
TranslogGeneration translogGeneration = config.getTranslogGeneration();
if (translogGeneration == null || translogGeneration.translogUUID == null) { // legacy case
if (translogGeneration == null || translogGeneration.translogUUID == null) { // legacy case
translogUUID = Strings.randomBase64UUID();
} else {
translogUUID = translogGeneration.translogUUID;
@ -190,6 +190,7 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC
boolean success = false;
ArrayList<ImmutableTranslogReader> foundTranslogs = new ArrayList<>();
final Path tempFile = Files.createTempFile(location, TRANSLOG_FILE_PREFIX, TRANSLOG_FILE_SUFFIX); // a temp file to copy checkpoint to - note it must be in on the same FS otherwise atomic move won't work
boolean tempFileRenamed = false;
try (ReleasableLock lock = writeLock.acquire()) {
logger.debug("open uncommitted translog checkpoint {}", checkpoint);
final String checkpointTranslogFile = getFilename(checkpoint.generation);
@ -215,6 +216,7 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC
Files.copy(location.resolve(CHECKPOINT_FILE_NAME), tempFile, StandardCopyOption.REPLACE_EXISTING);
IOUtils.fsync(tempFile, false);
Files.move(tempFile, commitCheckpoint, StandardCopyOption.ATOMIC_MOVE);
tempFileRenamed = true;
// we only fsync the directory the tempFile was already fsynced
IOUtils.fsync(commitCheckpoint.getParent(), true);
}
@ -223,10 +225,12 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC
if (success == false) {
IOUtils.closeWhileHandlingException(foundTranslogs);
}
try {
Files.delete(tempFile);
} catch (IOException ex) {
logger.warn("failed to delete temp file {}", ex, tempFile);
if (tempFileRenamed == false) {
try {
Files.delete(tempFile);
} catch (IOException ex) {
logger.warn("failed to delete temp file {}", ex, tempFile);
}
}
}
return foundTranslogs;
@ -347,7 +351,6 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC
}
TranslogWriter createWriter(long fileGeneration) throws IOException {
TranslogWriter newFile;
try {
@ -508,6 +511,7 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC
/**
* Ensures that the given location has be synced / written to the underlying storage.
*
* @return Returns <code>true</code> iff this call caused an actual sync operation otherwise <code>false</code>
*/
public boolean ensureSynced(Location location) throws IOException {
@ -749,13 +753,21 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
Location location = (Location) o;
if (generation != location.generation) return false;
if (translogLocation != location.translogLocation) return false;
if (generation != location.generation) {
return false;
}
if (translogLocation != location.translogLocation) {
return false;
}
return size == location.size;
}
@ -1089,7 +1101,7 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC
}
@Override
public Source getSource(){
public Source getSource() {
throw new IllegalStateException("trying to read doc source from delete operation");
}
@ -1198,7 +1210,7 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC
// to prevent this unfortunately.
in.mark(opSize);
in.skip(opSize-4);
in.skip(opSize - 4);
verifyChecksum(in);
in.reset();
}
@ -1250,7 +1262,7 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC
out.writeByte(op.opType().id());
op.writeTo(out);
long checksum = out.getChecksum();
out.writeInt((int)checksum);
out.writeInt((int) checksum);
}
/**