Skip to content

Commit 4a0feee

Browse files
authored
handle unexpected exception on success callback of translog upload (#12577)
* handle unexpected exception on success callback of translog upload Signed-off-by: Varun Bansal <bansvaru@amazon.com>
1 parent d0467b3 commit 4a0feee

File tree

2 files changed

+42
-3
lines changed

2 files changed

+42
-3
lines changed

server/src/main/java/org/opensearch/index/translog/transfer/FileTransferTracker.java

+12-3
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,8 @@
88

99
package org.opensearch.index.translog.transfer;
1010

11+
import org.apache.logging.log4j.Logger;
12+
import org.opensearch.common.logging.Loggers;
1113
import org.opensearch.core.index.shard.ShardId;
1214
import org.opensearch.index.remote.RemoteTranslogTransferTracker;
1315
import org.opensearch.index.translog.transfer.FileSnapshot.TransferFileSnapshot;
@@ -33,11 +35,13 @@ public class FileTransferTracker implements FileTransferListener {
3335
private final RemoteTranslogTransferTracker remoteTranslogTransferTracker;
3436
private Map<String, Long> bytesForTlogCkpFileToUpload;
3537
private long fileTransferStartTime = -1;
38+
private final Logger logger;
3639

3740
public FileTransferTracker(ShardId shardId, RemoteTranslogTransferTracker remoteTranslogTransferTracker) {
3841
this.shardId = shardId;
3942
this.fileTransferTracker = new ConcurrentHashMap<>();
4043
this.remoteTranslogTransferTracker = remoteTranslogTransferTracker;
44+
this.logger = Loggers.getLogger(getClass(), shardId);
4145
}
4246

4347
void recordFileTransferStartTime(long uploadStartTime) {
@@ -64,9 +68,14 @@ long getTotalBytesToUpload() {
6468

6569
@Override
6670
public void onSuccess(TransferFileSnapshot fileSnapshot) {
67-
long durationInMillis = (System.nanoTime() - fileTransferStartTime) / 1_000_000L;
68-
remoteTranslogTransferTracker.addUploadTimeInMillis(durationInMillis);
69-
remoteTranslogTransferTracker.addUploadBytesSucceeded(bytesForTlogCkpFileToUpload.get(fileSnapshot.getName()));
71+
try {
72+
long durationInMillis = (System.nanoTime() - fileTransferStartTime) / 1_000_000L;
73+
remoteTranslogTransferTracker.addUploadTimeInMillis(durationInMillis);
74+
remoteTranslogTransferTracker.addUploadBytesSucceeded(bytesForTlogCkpFileToUpload.get(fileSnapshot.getName()));
75+
} catch (Exception ex) {
76+
logger.error("Failure to update translog upload success stats", ex);
77+
}
78+
7079
add(fileSnapshot.getName(), TransferState.SUCCESS);
7180
}
7281

server/src/test/java/org/opensearch/index/translog/transfer/FileTransferTrackerTests.java

+30
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,10 @@
2020
import java.util.List;
2121
import java.util.Set;
2222

23+
import static org.mockito.Mockito.anyLong;
24+
import static org.mockito.Mockito.doAnswer;
25+
import static org.mockito.Mockito.spy;
26+
2327
public class FileTransferTrackerTests extends OpenSearchTestCase {
2428

2529
protected final ShardId shardId = new ShardId("index", "_na_", 1);
@@ -94,6 +98,32 @@ public void testOnFailure() throws IOException {
9498
}
9599
}
96100

101+
public void testOnSuccessStatsFailure() throws IOException {
102+
RemoteTranslogTransferTracker localRemoteTranslogTransferTracker = spy(remoteTranslogTransferTracker);
103+
doAnswer((count) -> { throw new NullPointerException("Error while updating stats"); }).when(localRemoteTranslogTransferTracker)
104+
.addUploadBytesSucceeded(anyLong());
105+
106+
FileTransferTracker localFileTransferTracker = new FileTransferTracker(shardId, localRemoteTranslogTransferTracker);
107+
108+
Path testFile = createTempFile();
109+
int fileSize = 128;
110+
Files.write(testFile, randomByteArrayOfLength(fileSize), StandardOpenOption.APPEND);
111+
try (
112+
FileSnapshot.TransferFileSnapshot transferFileSnapshot = new FileSnapshot.TransferFileSnapshot(
113+
testFile,
114+
randomNonNegativeLong(),
115+
null
116+
);
117+
) {
118+
Set<FileSnapshot.TransferFileSnapshot> toUpload = new HashSet<>(2);
119+
toUpload.add(transferFileSnapshot);
120+
localFileTransferTracker.recordBytesForFiles(toUpload);
121+
localRemoteTranslogTransferTracker.addUploadBytesStarted(fileSize);
122+
localFileTransferTracker.onSuccess(transferFileSnapshot);
123+
assertEquals(localFileTransferTracker.allUploaded().size(), 1);
124+
}
125+
}
126+
97127
public void testUploaded() throws IOException {
98128
Path testFile = createTempFile();
99129
int fileSize = 128;

0 commit comments

Comments
 (0)