honeymoose/OpenSearch
1
0
Fork 0
mirror of https://github.com/honeymoose/OpenSearch.git synced 2025-03-24 17:09:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

Translog Flush: When disabling flush and enabling it again, scheduled flush stops executing, closes #1727.

Browse Source
This commit is contained in:
Shay Banon 2012-02-22 15:00:58 +02:00
parent 31b793591e
commit 0ef2afb855
1 changed files with 35 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
src/main/java/org/elasticsearch/index/translog/TranslogService.java
View File

@ -26,11 +26,11 @@ import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue; import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.engine.Engine; import org.elasticsearch.index.engine.Engine;
import org.elasticsearch.index.engine.EngineClosedException;
import org.elasticsearch.index.engine.FlushNotAllowedEngineException; import org.elasticsearch.index.engine.FlushNotAllowedEngineException;
import org.elasticsearch.index.settings.IndexSettings; import org.elasticsearch.index.settings.IndexSettings;
import org.elasticsearch.index.settings.IndexSettingsService; import org.elasticsearch.index.settings.IndexSettingsService;
import org.elasticsearch.index.shard.AbstractIndexShardComponent; import org.elasticsearch.index.shard.AbstractIndexShardComponent;
import org.elasticsearch.index.shard.IllegalIndexShardStateException;
import org.elasticsearch.index.shard.IndexShardState; import org.elasticsearch.index.shard.IndexShardState;
import org.elasticsearch.index.shard.ShardId; import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.index.shard.service.IndexShard; import org.elasticsearch.index.shard.service.IndexShard;
@ -138,30 +138,48 @@ public class TranslogService extends AbstractIndexShardComponent {
if (indexShard.state() == IndexShardState.CLOSED) { if (indexShard.state() == IndexShardState.CLOSED) {
return; return;
} }
// flush is disabled, but still reschedule
if (disableFlush) { if (disableFlush) {
reschedule();
return; return;
} }
int currentNumberOfOperations = translog.estimatedNumberOfOperations(); if (indexShard.state() == IndexShardState.CREATED) {
if (currentNumberOfOperations > flushThresholdOperations) { reschedule();
logger.trace("flushing translog, operations [{}], breached [{}]", currentNumberOfOperations, flushThresholdOperations);
asyncFlushAndReschedule();
return; return;
} }
long sizeInBytes = translog.translogSizeInBytes(); if (flushThresholdOperations > 0) {
if (sizeInBytes > flushThresholdSize.bytes()) { int currentNumberOfOperations = translog.estimatedNumberOfOperations();
logger.trace("flushing translog, size [{}], breached [{}]", new ByteSizeValue(sizeInBytes), flushThresholdSize); if (currentNumberOfOperations > flushThresholdOperations) {
asyncFlushAndReschedule(); logger.trace("flushing translog, operations [{}], breached [{}]", currentNumberOfOperations, flushThresholdOperations);
return; asyncFlushAndReschedule();
return;
}
} }
if ((System.currentTimeMillis() - lastFlushTime) > flushThresholdPeriod.millis()) { if (flushThresholdSize.bytes() > 0) {
logger.trace("flushing translog, last_flush_time [{}], breached [{}]", lastFlushTime, flushThresholdPeriod); long sizeInBytes = translog.translogSizeInBytes();
asyncFlushAndReschedule(); if (sizeInBytes > flushThresholdSize.bytes()) {
return; logger.trace("flushing translog, size [{}], breached [{}]", new ByteSizeValue(sizeInBytes), flushThresholdSize);
asyncFlushAndReschedule();
return;
}
} }
if (flushThresholdPeriod.millis() > 0) {
if ((threadPool.estimatedTimeInMillis() - lastFlushTime) > flushThresholdPeriod.millis()) {
logger.trace("flushing translog, last_flush_time [{}], breached [{}]", lastFlushTime, flushThresholdPeriod);
asyncFlushAndReschedule();
return;
}
}
reschedule();
}
private void reschedule() {
future = threadPool.schedule(interval, ThreadPool.Names.SAME, this); future = threadPool.schedule(interval, ThreadPool.Names.SAME, this);
} }
@ -171,14 +189,14 @@ public class TranslogService extends AbstractIndexShardComponent {
public void run() { public void run() {
try { try {
indexShard.flush(new Engine.Flush()); indexShard.flush(new Engine.Flush());
} catch (EngineClosedException e) { } catch (IllegalIndexShardStateException e) {
// we are being closed, ignore // we are being closed, or in created state, ignore
} catch (FlushNotAllowedEngineException e) { } catch (FlushNotAllowedEngineException e) {
// ignore this exception, we are not allowed to perform flush // ignore this exception, we are not allowed to perform flush
} catch (Exception e) { } catch (Exception e) {
logger.warn("failed to flush shard on translog threshold", e); logger.warn("failed to flush shard on translog threshold", e);
} }
lastFlushTime = System.currentTimeMillis(); lastFlushTime = threadPool.estimatedTimeInMillis();
if (indexShard.state() != IndexShardState.CLOSED) { if (indexShard.state() != IndexShardState.CLOSED) {
future = threadPool.schedule(interval, ThreadPool.Names.SAME, TranslogBasedFlush.this); future = threadPool.schedule(interval, ThreadPool.Names.SAME, TranslogBasedFlush.this);