honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

add setters for translog location and took in engine operation result

This commit is contained in:
Areek Zillur 2016-10-25 09:58:14 -04:00
parent bb785483ae
commit 64a897e5f2
6 changed files with 75 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
core/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java
View File

@ -243,7 +243,7 @@ public class TransportShardBulkAction extends TransportWriteAction<BulkShardRequ
} catch (Exception failure) {
// we may fail translating a update to index or delete operation
// we use index result to communicate failure while translating update request
updateOperationResult = new Engine.IndexResult(failure, updateRequest.version(), 0, 0);
updateOperationResult = new Engine.IndexResult(failure, updateRequest.version(), 0);
break; // out of retry loop
}
// execute translated update request

8
core/src/main/java/org/elasticsearch/action/index/TransportIndexAction.java
View File

@ -167,7 +167,7 @@ public class TransportIndexAction extends TransportWriteAction<IndexRequest, Ind
try {
operation = replica.prepareIndexOnReplica(sourceToParse, request.version(), request.versionType(), request.getAutoGeneratedTimestamp(), request.isRetry());
} catch (MapperParsingException | IllegalArgumentException e) {
return new Engine.IndexResult(e, request.version(), 0, 0);
return new Engine.IndexResult(e, request.version(), 0);
}
Mapping update = operation.parsedDoc().dynamicMappingsUpdate();
if (update != null) {
@ -189,7 +189,7 @@ public class TransportIndexAction extends TransportWriteAction<IndexRequest, Ind
try {
operation = prepareIndexOperationOnPrimary(request, primary);
} catch (MapperParsingException | IllegalArgumentException e) {
return new Engine.IndexResult(e, request.version(), 0, 0);
return new Engine.IndexResult(e, request.version(), 0);
}
Mapping update = operation.parsedDoc().dynamicMappingsUpdate();
final ShardId shardId = primary.shardId();
@ -201,13 +201,13 @@ public class TransportIndexAction extends TransportWriteAction<IndexRequest, Ind
} catch (IllegalArgumentException e) {
// throws IAE on conflicts merging dynamic mappings
return new Engine.IndexResult(e, request.version(),
operation.startTime() - System.nanoTime(), 0);
0);
}
try {
operation = prepareIndexOperationOnPrimary(request, primary);
} catch (MapperParsingException | IllegalArgumentException e) {
return new Engine.IndexResult(e, request.version(),
operation.startTime() - System.nanoTime(), 0);
0);
}
update = operation.parsedDoc().dynamicMappingsUpdate();
if (update != null) {

57
core/src/main/java/org/elasticsearch/index/engine/Engine.java
View File

@ -284,30 +284,23 @@ public abstract class Engine implements Closeable {
public abstract static class Result {
private final Operation.TYPE operationType;
private final Translog.Location location;
private final long version;
private final Exception failure;
private final long took;
private final int estimatedSizeInBytes;
private Translog.Location location;
private long took;
private boolean freeze;
private Result(Operation.TYPE operationType, Translog.Location location, Exception failure,
long version, long took, int estimatedSizeInBytes) {
protected Result(Operation.TYPE operationType, Exception failure,
long version, int estimatedSizeInBytes) {
this.operationType = operationType;
this.location = location;
this.failure = failure;
this.version = version;
this.took = took;
this.estimatedSizeInBytes = estimatedSizeInBytes;
}
protected Result(Operation.TYPE operationType, Translog.Location location,
long version, long took, int estimatedSizeInBytes) {
this(operationType, location, null, version, took, estimatedSizeInBytes);
}
protected Result(Operation.TYPE operationType, Exception failure,
long version, long took, int estimatedSizeInBytes) {
this(operationType, null, failure, version, took, estimatedSizeInBytes);
protected Result(Operation.TYPE operationType, long version, int estimatedSizeInBytes) {
this(operationType, null, version, estimatedSizeInBytes);
}
public boolean hasFailure() {
@ -340,18 +333,38 @@ public abstract class Engine implements Closeable {
}
return estimatedSizeInBytes;
}
public void setLocation(Translog.Location location) {
if (freeze == false) {
this.location = location;
} else {
throw new IllegalStateException("result is already frozen");
}
}
public void setTook(long took) {
if (freeze == false) {
this.took = took;
} else {
throw new IllegalStateException("result is already frozen");
}
}
public void freeze() {
this.freeze = true;
}
}
public static class IndexResult extends Result {
private final boolean created;
public IndexResult(Translog.Location location, long version, boolean created, long took, int estimatedSizeInBytes) {
super(Operation.TYPE.INDEX, location, version, took, estimatedSizeInBytes);
public IndexResult(long version, boolean created, int estimatedSizeInBytes) {
super(Operation.TYPE.INDEX, version, estimatedSizeInBytes);
this.created = created;
}
public IndexResult(Exception failure, long version, long took, int estimatedSizeInBytes) {
super(Operation.TYPE.INDEX, failure, version, took, estimatedSizeInBytes);
public IndexResult(Exception failure, long version, int estimatedSizeInBytes) {
super(Operation.TYPE.INDEX, failure, version, estimatedSizeInBytes);
this.created = false;
}
@ -363,13 +376,13 @@ public abstract class Engine implements Closeable {
public static class DeleteResult extends Result {
private final boolean found;
public DeleteResult(Translog.Location location, long version, boolean found, long took, int estimatedSizeInBytes) {
super(Operation.TYPE.DELETE, location, version, took, estimatedSizeInBytes);
public DeleteResult(long version, boolean found, int estimatedSizeInBytes) {
super(Operation.TYPE.DELETE, version, estimatedSizeInBytes);
this.found = found;
}
DeleteResult(Exception failure, long version, long took, int estimatedSizeInBytes) {
super(Operation.TYPE.DELETE, failure, version, took, estimatedSizeInBytes);
public DeleteResult(Exception failure, long version, int estimatedSizeInBytes) {
super(Operation.TYPE.DELETE, failure, version, estimatedSizeInBytes);
this.found = false;
}

59
core/src/main/java/org/elasticsearch/index/engine/InternalEngine.java
View File

@ -387,26 +387,6 @@ public class InternalEngine extends Engine {
return currentVersion;
}
private static VersionValueSupplier NEW_VERSION_VALUE = (u, t) -> new VersionValue(u);
@FunctionalInterface
private interface VersionValueSupplier {
VersionValue apply(long updatedVersion, long time);
}
private <T extends Engine.Operation> Translog.Location maybeAddToTranslog(
final T op,
final long updatedVersion,
final Function<T, Translog.Operation> toTranslogOp,
final VersionValueSupplier toVersionValue) throws IOException {
Translog.Location location = null;
if (op.origin() != Operation.Origin.LOCAL_TRANSLOG_RECOVERY) {
location = translog.add(toTranslogOp.apply(op));
}
versionMap.putUnderLock(op.uid().bytes(), toVersionValue.apply(updatedVersion, engineConfig.getThreadPool().estimatedTimeInMillis()));
return location;
}
@Override
public IndexResult index(Index index) {
IndexResult result;
@ -423,7 +403,7 @@ public class InternalEngine extends Engine {
} catch (Exception e) {
Exception documentFailure = extractDocumentFailure(index, e);
result = new IndexResult(documentFailure, index.version(),
index.startTime() - System.nanoTime(), index.estimatedSizeInBytes());
index.estimatedSizeInBytes());
}
return result;
}
@ -551,7 +531,7 @@ public class InternalEngine extends Engine {
final long expectedVersion = index.version();
if (checkVersionConflict(index, currentVersion, expectedVersion, deleted)) {
// skip index operation because of version conflict on recovery
return new IndexResult(null, expectedVersion, false, index.startTime() - System.nanoTime(), index.estimatedSizeInBytes());
return new IndexResult(expectedVersion, false, index.estimatedSizeInBytes());
} else {
updatedVersion = index.versionType().updateVersion(currentVersion, expectedVersion);
index.parsedDoc().version().setLongValue(updatedVersion);
@ -561,8 +541,17 @@ public class InternalEngine extends Engine {
} else {
update(index.uid(), index.docs(), indexWriter);
}
location = maybeAddToTranslog(index, updatedVersion, Translog.Index::new, NEW_VERSION_VALUE);
return new IndexResult(location, updatedVersion, deleted, index.startTime() - System.nanoTime(), index.estimatedSizeInBytes());
IndexResult indexResult = new IndexResult(updatedVersion, deleted, index.estimatedSizeInBytes());
if (index.origin() != Operation.Origin.LOCAL_TRANSLOG_RECOVERY) {
location = translog.add(new Translog.Index(index, indexResult));
} else {
location = null;
}
versionMap.putUnderLock(index.uid().bytes(), new VersionValue(updatedVersion));
indexResult.setLocation(location);
indexResult.setTook(index.startTime() - System.nanoTime());
indexResult.freeze();
return indexResult;
}
}
}
@ -593,7 +582,7 @@ public class InternalEngine extends Engine {
} catch (Exception e) {
Exception documentFailure = extractDocumentFailure(delete, e);
result = new DeleteResult(documentFailure, delete.version(),
delete.startTime() - System.nanoTime(), delete.estimatedSizeInBytes());
delete.estimatedSizeInBytes());
}
maybePruneDeletedTombstones();
return result;
@ -628,14 +617,24 @@ public class InternalEngine extends Engine {
final long expectedVersion = delete.version();
if (checkVersionConflict(delete, currentVersion, expectedVersion, deleted)) {
// skip executing delete because of version conflict on recovery
return new DeleteResult(null, expectedVersion, true,
delete.startTime() - System.nanoTime(), delete.estimatedSizeInBytes());
return new DeleteResult(expectedVersion, true,
delete.estimatedSizeInBytes());
} else {
updatedVersion = delete.versionType().updateVersion(currentVersion, expectedVersion);
found = deleteIfFound(delete.uid(), currentVersion, deleted, versionValue);
location = maybeAddToTranslog(delete, updatedVersion, Translog.Delete::new, DeleteVersionValue::new);
return new DeleteResult(location, updatedVersion, found,
delete.startTime() - System.nanoTime(), delete.estimatedSizeInBytes());
DeleteResult deleteResult = new DeleteResult(updatedVersion, found,
delete.estimatedSizeInBytes());
if (delete.origin() != Operation.Origin.LOCAL_TRANSLOG_RECOVERY) {
location = translog.add(new Translog.Delete(delete, deleteResult));
} else {
location = null;
}
versionMap.putUnderLock(delete.uid().bytes(),
new DeleteVersionValue(updatedVersion, engineConfig.getThreadPool().estimatedTimeInMillis()));
deleteResult.setLocation(location);
deleteResult.setTook(delete.startTime() - System.nanoTime());
deleteResult.freeze();
return deleteResult;
}
}
}

8
core/src/main/java/org/elasticsearch/index/translog/Translog.java
View File

@ -830,13 +830,13 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC
}
}
public Index(Engine.Index index) {
public Index(Engine.Index index, Engine.IndexResult indexResult) {
this.id = index.id();
this.type = index.type();
this.source = index.source();
this.routing = index.routing();
this.parent = index.parent();
this.version = index.version();
this.version = indexResult.getVersion();
this.timestamp = index.timestamp();
this.ttl = index.ttl();
this.versionType = index.versionType();
@ -994,9 +994,9 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC
assert versionType.validateVersionForWrites(this.version);
}
public Delete(Engine.Delete delete) {
public Delete(Engine.Delete delete, Engine.DeleteResult deleteResult) {
this.uid = delete.uid();
this.version = delete.version();
this.version = deleteResult.getVersion();
this.versionType = delete.versionType();
}

4
core/src/test/java/org/elasticsearch/index/shard/IndexingOperationListenerTests.java
View File

@ -112,7 +112,7 @@ public class IndexingOperationListenerTests extends ESTestCase{
IndexingOperationListener.CompositeListener compositeListener = new IndexingOperationListener.CompositeListener(indexingOperationListeners, logger);
Engine.Delete delete = new Engine.Delete("test", "1", new Term("_uid", "1"));
Engine.Index index = new Engine.Index(new Term("_uid", "1"), null);
compositeListener.postDelete(delete, new Engine.DeleteResult(null, 1, true, 0, 0));
compositeListener.postDelete(delete, new Engine.DeleteResult(1, true, 0));
assertEquals(0, preIndex.get());
assertEquals(0, postIndex.get());
assertEquals(0, postIndexException.get());
@ -136,7 +136,7 @@ public class IndexingOperationListenerTests extends ESTestCase{
assertEquals(2, postDelete.get());
assertEquals(2, postDeleteException.get());
compositeListener.postIndex(index, new Engine.IndexResult(null, 0, false, 0, 0));
compositeListener.postIndex(index, new Engine.IndexResult(0, false, 0));
assertEquals(0, preIndex.get());
assertEquals(2, postIndex.get());
assertEquals(0, postIndexException.get());