Skip to content

Commit

Permalink
rename
Browse files Browse the repository at this point in the history
  • Loading branch information
martijnvg committed Oct 24, 2018
1 parent 7af8ba2 commit 9925984
Showing 1 changed file with 6 additions and 6 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ public abstract class ShardFollowNodeTask extends AllocatedPersistentTask {
private long numberOfOperationsIndexed = 0;
private long lastFetchTime = -1;
private final Queue<Translog.Operation> buffer = new PriorityQueue<>(Comparator.comparing(Translog.Operation::seqNo));
private long bufferSize = 0;
private long bufferSizeInBytes = 0;
private final LinkedHashMap<Long, Tuple<AtomicInteger, ElasticsearchException>> fetchExceptions;

private volatile ElasticsearchException fatalException;
Expand Down Expand Up @@ -184,8 +184,8 @@ private boolean hasReadBudget() {
params.getFollowShardId(), numConcurrentReads);
return false;
}
if (bufferSize >= params.getMaxWriteBufferSize().getBytes()) {
LOGGER.trace("{} no new reads, buffer size limit has been reached [{}]", params.getFollowShardId(), bufferSize);
if (bufferSizeInBytes >= params.getMaxWriteBufferSize().getBytes()) {
LOGGER.trace("{} no new reads, buffer size limit has been reached [{}]", params.getFollowShardId(), bufferSizeInBytes);
return false;
}
if (buffer.size() > params.getMaxWriteBufferCount()) {
Expand Down Expand Up @@ -214,7 +214,7 @@ private synchronized void coordinateWrites() {
}
}
long opsSize = ops.stream().mapToLong(Translog.Operation::estimateSize).sum();
bufferSize -= opsSize;
bufferSizeInBytes -= opsSize;
numConcurrentWrites++;
LOGGER.trace("{}[{}] write [{}/{}] [{}]", params.getFollowShardId(), numConcurrentWrites, ops.get(0).seqNo(),
ops.get(ops.size() - 1).seqNo(), ops.size());
Expand Down Expand Up @@ -293,7 +293,7 @@ synchronized void innerHandleReadResponse(long from, long maxRequiredSeqNo, Shar
.mapToLong(Translog.Operation::estimateSize)
.sum();
buffer.addAll(operations);
bufferSize += operationsSize;
bufferSizeInBytes += operationsSize;
final long maxSeqNo = response.getOperations()[response.getOperations().length - 1].seqNo();
assert maxSeqNo ==
Arrays.stream(response.getOperations()).mapToLong(Translog.Operation::seqNo).max().getAsLong();
Expand Down Expand Up @@ -467,7 +467,7 @@ public synchronized ShardFollowNodeTaskStatus getStatus() {
numConcurrentReads,
numConcurrentWrites,
buffer.size(),
bufferSize,
bufferSizeInBytes,
currentMappingVersion,
totalFetchTimeMillis,
totalFetchTookTimeMillis,
Expand Down

0 comments on commit 9925984

Please sign in to comment.