Skip to content
New issue

Have a question about this project? # for a free GitHub account to open an issue and contact its maintainers and the community.

#

By clicking “#”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? # to your account

fix(s3stream): fix kos issue403 repeated add backoff record #498

Merged
merged 1 commit into from
Oct 29, 2023
Merged

fix(s3stream): fix kos issue403 repeated add backoff record #498

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions s3stream/src/main/java/com/automq/stream/s3/S3Storage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -241,7 +241,9 @@ public boolean append0(WalWriteRequest request, boolean fromBackoff) {
return true;
}
if (!tryAcquirePermit()) {
backoffRecords.offer(request);
if (!fromBackoff) {
backoffRecords.offer(request);
}
OperationMetricsStats.getOrCreateOperationMetrics(S3Operation.APPEND_STORAGE_LOG_CACHE_FULL).operationCount.inc();
if (System.currentTimeMillis() - lastLogTimestamp > 1000L) {
LOGGER.warn("[BACKOFF] log cache size {} is larger than {}", logCache.size(), maxWALCacheSize);
Expand All @@ -257,7 +259,9 @@ public boolean append0(WalWriteRequest request, boolean fromBackoff) {
} catch (WriteAheadLog.OverCapacityException e) {
// the WAL write data align with block, 'WAL is full but LogCacheBlock is not full' may happen.
forceUpload(LogCache.MATCH_ALL_STREAMS);
backoffRecords.offer(request);
if (!fromBackoff) {
backoffRecords.offer(request);
}
if (System.currentTimeMillis() - lastLogTimestamp > 1000L) {
LOGGER.warn("[BACKOFF] log over capacity", e);
lastLogTimestamp = System.currentTimeMillis();
Expand Down
1 change: 1 addition & 0 deletions s3stream/src/main/java/com/automq/stream/s3/compact/CompactionUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,7 @@ public static Map<Long, List<StreamDataBlock>> blockWaitObjectIndices(List<Strea
return new AbstractMap.SimpleEntry<>(f.getKey(), validStreamDataBlocks);
} catch (Exception ex) {
// continue compaction without invalid object
// TODO: log warn
return null;
}
})
Expand Down
5 changes: 2 additions & 3 deletions s3stream/src/main/java/com/automq/stream/s3/compact/operator/DataBlockReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@
package com.automq.stream.s3.compact.operator;

import com.automq.stream.s3.DirectByteBufAlloc;
import com.automq.stream.s3.ObjectReader;
import com.automq.stream.s3.compact.objects.StreamDataBlock;
import com.automq.stream.s3.network.ThrottleStrategy;
import com.automq.stream.s3.operator.S3Operator;
Expand All @@ -36,7 +35,7 @@

//TODO: refactor to reduce duplicate code with ObjectWriter
public class DataBlockReader {
private static final Logger LOGGER = LoggerFactory.getLogger(ObjectReader.class);
private static final Logger LOGGER = LoggerFactory.getLogger(DataBlockReader.class);
private final S3ObjectMetadata metadata;
private final String objectKey;
private final S3Operator s3Operator;
Expand Down Expand Up @@ -66,7 +65,7 @@ public void parseDataBlockIndex(long startPosition) {
}
}).exceptionally(ex -> {
// unrecoverable error, possibly read on a deleted object
LOGGER.warn("s3 range read from {} [{}, {}) failed, ex", objectKey, startPosition, metadata.objectSize(), ex);
LOGGER.warn("s3 range read from {} [{}, {}) failed", objectKey, startPosition, metadata.objectSize(), ex);
indexBlockCf.completeExceptionally(ex);
return null;
});
Expand Down
4 changes: 2 additions & 2 deletions s3stream/src/main/java/com/automq/stream/s3/operator/DefaultS3Operator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,10 +243,10 @@ void mergedRangeRead0(String path, long start, long end, CompletableFuture<ByteB
OperationMetricsStats.getOrCreateOperationMetrics(S3Operation.GET_OBJECT_FAIL).operationCount.inc();
OperationMetricsStats.getOrCreateOperationMetrics(S3Operation.GET_OBJECT_FAIL).operationTime.update(timerUtil.elapsed());
if (isUnrecoverable(ex)) {
LOGGER.error("GetObject for object {} [{}, {})fail", path, start, end, ex);
LOGGER.error("GetObject for object {} [{}, {}) fail", path, start, end, ex);
cf.completeExceptionally(ex);
} else {
LOGGER.warn("GetObject for object {} [{}, {})fail, retry later", path, start, end, ex);
LOGGER.warn("GetObject for object {} [{}, {}) fail, retry later", path, start, end, ex);
scheduler.schedule(() -> mergedRangeRead0(path, start, end, cf), 100, TimeUnit.MILLISECONDS);
}
return null;
Expand Down