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

Move `UNASSIGNED_SEQ_NO` and `NO_OPS_PERFORMED` to SequenceNumbers (#26494) 

Where they better belong.
This commit is contained in:
Boaz Leskes 2017-09-04 16:31:00 +02:00 committed by GitHub
parent a4d5c6418e
commit 2fd4af82e4
42 changed files with 336 additions and 338 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -33,7 +33,7 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.index.Index;
import org.elasticsearch.index.IndexSettings;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.rest.RestStatus;
@ -176,7 +176,7 @@ public abstract class DocWriteResponse extends ReplicationResponse implements Wr
}
/**
* Returns the sequence number assigned for this change. Returns {@link SequenceNumbersService#UNASSIGNED_SEQ_NO} if the operation
* Returns the sequence number assigned for this change. Returns {@link SequenceNumbers#UNASSIGNED_SEQ_NO} if the operation
* wasn't performed (i.e., an update operation that resulted in a NOOP).
*/
public long getSeqNo() {
@ -263,7 +263,7 @@ public abstract class DocWriteResponse extends ReplicationResponse implements Wr
seqNo = in.readZLong();
primaryTerm = in.readVLong();
} else {
seqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO;
seqNo = SequenceNumbers.UNASSIGNED_SEQ_NO;
primaryTerm = 0;
}
forcedRefresh = in.readBoolean();
@ -375,7 +375,7 @@ public abstract class DocWriteResponse extends ReplicationResponse implements Wr
protected Result result = null;
protected boolean forcedRefresh;
protected ShardInfo shardInfo = null;
protected Long seqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO;
protected Long seqNo = SequenceNumbers.UNASSIGNED_SEQ_NO;
protected Long primaryTerm = 0L;
public ShardId getShardId() {

12
core/src/main/java/org/elasticsearch/action/bulk/BulkItemResponse.java
View File

@ -37,7 +37,7 @@ import org.elasticsearch.common.xcontent.StatusToXContentObject;
import org.elasticsearch.common.xcontent.ToXContentFragment;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.rest.RestStatus;
import java.io.IOException;
@ -182,15 +182,15 @@ public class BulkItemResponse implements Streamable, StatusToXContentObject {
* to record operation sequence no with failure
*/
public Failure(String index, String type, String id, Exception cause) {
this(index, type, id, cause, ExceptionsHelper.status(cause), SequenceNumbersService.UNASSIGNED_SEQ_NO, false);
this(index, type, id, cause, ExceptionsHelper.status(cause), SequenceNumbers.UNASSIGNED_SEQ_NO, false);
}
public Failure(String index, String type, String id, Exception cause, boolean aborted) {
this(index, type, id, cause, ExceptionsHelper.status(cause), SequenceNumbersService.UNASSIGNED_SEQ_NO, aborted);
this(index, type, id, cause, ExceptionsHelper.status(cause), SequenceNumbers.UNASSIGNED_SEQ_NO, aborted);
}
public Failure(String index, String type, String id, Exception cause, RestStatus status) {
this(index, type, id, cause, status, SequenceNumbersService.UNASSIGNED_SEQ_NO, false);
this(index, type, id, cause, status, SequenceNumbers.UNASSIGNED_SEQ_NO, false);
}
/** For write failures after operation was assigned a sequence number. */
@ -220,7 +220,7 @@ public class BulkItemResponse implements Streamable, StatusToXContentObject {
if (in.getVersion().onOrAfter(Version.V_6_0_0_alpha1)) {
seqNo = in.readZLong();
} else {
seqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO;
seqNo = SequenceNumbers.UNASSIGNED_SEQ_NO;
}
if (supportsAbortedFlag(in.getVersion())) {
aborted = in.readBoolean();
@ -292,7 +292,7 @@ public class BulkItemResponse implements Streamable, StatusToXContentObject {
/**
* The operation sequence number generated by primary
* NOTE: {@link SequenceNumbersService#UNASSIGNED_SEQ_NO}
* NOTE: {@link SequenceNumbers#UNASSIGNED_SEQ_NO}
* indicates sequence number was not generated by primary
*/
public long getSeqNo() {

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

@ -56,7 +56,7 @@ import org.elasticsearch.index.engine.VersionConflictEngineException;
import org.elasticsearch.index.get.GetResult;
import org.elasticsearch.index.mapper.Mapping;
import org.elasticsearch.index.mapper.SourceToParse;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.shard.IndexShard;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.index.translog.Translog;
@ -275,7 +275,7 @@ public class TransportShardBulkAction extends TransportWriteAction<BulkShardRequ
static BulkItemResultHolder processUpdateResponse(final UpdateRequest updateRequest, final String concreteIndex,
final Engine.Result result, final UpdateHelper.Result translate,
final IndexShard primary, final int bulkReqId) throws Exception {
assert result.getSeqNo() != SequenceNumbersService.UNASSIGNED_SEQ_NO : "failed result should not have a sequence number";
assert result.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO : "failed result should not have a sequence number";
Engine.Operation.TYPE opType = result.getOperationType();
@ -344,7 +344,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
final Engine.Result result = new Engine.IndexResult(failure, updateRequest.version(), SequenceNumbersService.UNASSIGNED_SEQ_NO);
final Engine.Result result = new Engine.IndexResult(failure, updateRequest.version(), SequenceNumbers.UNASSIGNED_SEQ_NO);
return new BulkItemResultHolder(null, result, primaryItemRequest);
}
@ -446,7 +446,7 @@ public class TransportShardBulkAction extends TransportWriteAction<BulkShardRequ
final BulkItemResponse primaryResponse = request.getPrimaryResponse();
assert primaryResponse != null : "expected primary response to be set for item [" + index + "] request [" + request.request() + "]";
if (primaryResponse.isFailed()) {
return primaryResponse.getFailure().getSeqNo() != SequenceNumbersService.UNASSIGNED_SEQ_NO
return primaryResponse.getFailure().getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO
? ReplicaItemExecutionMode.FAILURE // we have a seq no generated with the failure, replicate as no-op
: ReplicaItemExecutionMode.NOOP; // no seq no generated, ignore replication
} else {
@ -485,7 +485,7 @@ public class TransportShardBulkAction extends TransportWriteAction<BulkShardRequ
break;
case FAILURE:
final BulkItemResponse.Failure failure = item.getPrimaryResponse().getFailure();
assert failure.getSeqNo() != SequenceNumbersService.UNASSIGNED_SEQ_NO : "seq no must be assigned";
assert failure.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO : "seq no must be assigned";
operationResult = replica.markSeqNoAsNoop(failure.getSeqNo(), failure.getMessage());
assert operationResult != null : "operation result must never be null when primary response has no failure";
location = syncOperationResultOrThrow(operationResult, location);

5
core/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java
View File

@ -53,6 +53,7 @@ import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.util.concurrent.AbstractRunnable;
import org.elasticsearch.index.IndexNotFoundException;
import org.elasticsearch.index.IndexService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.shard.IndexShard;
import org.elasticsearch.index.shard.IndexShardState;
@ -1035,7 +1036,7 @@ public abstract class TransportReplicationAction<
* checkpoint value when simulating responses to replication actions that pre-6.0 nodes are not aware of (e.g., the global
* checkpoint background sync, and the primary/replica resync).
*/
assert localCheckpoint != SequenceNumbersService.UNASSIGNED_SEQ_NO;
assert localCheckpoint != SequenceNumbers.UNASSIGNED_SEQ_NO;
this.localCheckpoint = localCheckpoint;
}
@ -1235,7 +1236,7 @@ public abstract class TransportReplicationAction<
if (in.getVersion().onOrAfter(Version.V_6_0_0_alpha1)) {
globalCheckpoint = in.readZLong();
} else {
globalCheckpoint = SequenceNumbersService.UNASSIGNED_SEQ_NO;
globalCheckpoint = SequenceNumbers.UNASSIGNED_SEQ_NO;
}
}

4
core/src/main/java/org/elasticsearch/action/update/UpdateResponse.java
View File

@ -25,7 +25,7 @@ import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.index.get.GetResult;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.rest.RestStatus;
@ -47,7 +47,7 @@ public class UpdateResponse extends DocWriteResponse {
* For example: update script with operation set to none
*/
public UpdateResponse(ShardId shardId, String type, String id, long version, Result result) {
this(new ShardInfo(0, 0), shardId, type, id, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, version, result);
this(new ShardInfo(0, 0), shardId, type, id, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, version, result);
}
public UpdateResponse(

4
core/src/main/java/org/elasticsearch/common/lucene/uid/PerThreadIDVersionAndSeqNoLookup.java
View File

@ -32,7 +32,7 @@ import org.elasticsearch.common.lucene.uid.VersionsAndSeqNoResolver.DocIdAndSeqN
import org.elasticsearch.common.lucene.uid.VersionsAndSeqNoResolver.DocIdAndVersion;
import org.elasticsearch.index.mapper.SeqNoFieldMapper;
import org.elasticsearch.index.mapper.VersionFieldMapper;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import java.io.IOException;
@ -138,7 +138,7 @@ final class PerThreadIDVersionAndSeqNoLookup {
if (seqNos != null && seqNos.advanceExact(docID)) {
seqNo = seqNos.longValue();
} else {
seqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO;
seqNo = SequenceNumbers.UNASSIGNED_SEQ_NO;
}
return new DocIdAndSeqNo(docID, seqNo, context);
} else {

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

@ -65,6 +65,7 @@ import org.elasticsearch.index.VersionType;
import org.elasticsearch.index.mapper.ParseContext.Document;
import org.elasticsearch.index.mapper.ParsedDocument;
import org.elasticsearch.index.merge.MergeStats;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.index.store.Store;
@ -396,7 +397,7 @@ public abstract class Engine implements Closeable {
* (e.g while preparing operation or updating mappings)
* */
public IndexResult(Exception failure, long version) {
this(failure, version, SequenceNumbersService.UNASSIGNED_SEQ_NO);
this(failure, version, SequenceNumbers.UNASSIGNED_SEQ_NO);
}
public IndexResult(Exception failure, long version, long seqNo) {
@ -1046,7 +1047,7 @@ public abstract class Engine implements Closeable {
Index(Term uid, ParsedDocument doc, long version) {
// use a primary term of 2 to allow tests to reduce it to a valid >0 term
this(uid, doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 2, version, VersionType.INTERNAL,
this(uid, doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 2, version, VersionType.INTERNAL,
Origin.PRIMARY, System.nanoTime(), -1, false);
} // TEST ONLY
@ -1122,7 +1123,7 @@ public abstract class Engine implements Closeable {
}
public Delete(String type, String id, Term uid) {
this(type, id, uid, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_ANY, VersionType.INTERNAL, Origin.PRIMARY, System.nanoTime());
this(type, id, uid, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_ANY, VersionType.INTERNAL, Origin.PRIMARY, System.nanoTime());
}
public Delete(Delete template, VersionType versionType) {

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

@ -179,14 +179,14 @@ public class InternalEngine extends Engine {
break;
case OPEN_INDEX_CREATE_TRANSLOG:
writer = createWriter(false);
seqNoStats = store.loadSeqNoStats(SequenceNumbersService.UNASSIGNED_SEQ_NO);
seqNoStats = store.loadSeqNoStats(SequenceNumbers.UNASSIGNED_SEQ_NO);
break;
case CREATE_INDEX_AND_TRANSLOG:
writer = createWriter(true);
seqNoStats = new SeqNoStats(
SequenceNumbersService.NO_OPS_PERFORMED,
SequenceNumbersService.NO_OPS_PERFORMED,
SequenceNumbersService.UNASSIGNED_SEQ_NO);
SequenceNumbers.NO_OPS_PERFORMED,
SequenceNumbers.NO_OPS_PERFORMED,
SequenceNumbers.UNASSIGNED_SEQ_NO);
break;
default:
throw new IllegalArgumentException(openMode.toString());
@ -463,7 +463,7 @@ public class InternalEngine extends Engine {
}
private OpVsLuceneDocStatus compareOpToLuceneDocBasedOnSeqNo(final Operation op) throws IOException {
assert op.seqNo() != SequenceNumbersService.UNASSIGNED_SEQ_NO : "resolving ops based on seq# but no seqNo is found";
assert op.seqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO : "resolving ops based on seq# but no seqNo is found";
final OpVsLuceneDocStatus status;
final VersionValue versionValue = versionMap.getUnderLock(op.uid());
assert incrementVersionLookup();
@ -507,7 +507,7 @@ public class InternalEngine extends Engine {
assert incrementIndexVersionLookup(); // used for asserting in tests
final long currentVersion = loadCurrentVersionFromIndex(op.uid());
if (currentVersion != Versions.NOT_FOUND) {
versionValue = new VersionValue(currentVersion, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0L);
versionValue = new VersionValue(currentVersion, SequenceNumbers.UNASSIGNED_SEQ_NO, 0L);
}
} else if (engineConfig.isEnableGcDeletes() && versionValue.isDelete() &&
(engineConfig.getThreadPool().relativeTimeInMillis() - ((DeleteVersionValue)versionValue).time) > getGcDeletesInMillis()) {
@ -518,7 +518,7 @@ public class InternalEngine extends Engine {
private OpVsLuceneDocStatus compareOpToLuceneDocBasedOnVersions(final Operation op)
throws IOException {
assert op.seqNo() == SequenceNumbersService.UNASSIGNED_SEQ_NO : "op is resolved based on versions but have a seq#";
assert op.seqNo() == SequenceNumbers.UNASSIGNED_SEQ_NO : "op is resolved based on versions but have a seq#";
assert op.version() >= 0 : "versions should be non-negative. got " + op.version();
final VersionValue versionValue = resolveDocVersion(op);
if (versionValue == null) {
@ -570,11 +570,11 @@ public class InternalEngine extends Engine {
private boolean assertIncomingSequenceNumber(final Engine.Operation.Origin origin, final long seqNo) {
if (engineConfig.getIndexSettings().getIndexVersionCreated().before(Version.V_6_0_0_alpha1) && origin == Operation.Origin.LOCAL_TRANSLOG_RECOVERY) {
// legacy support
assert seqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO : "old op recovering but it already has a seq no.;" +
assert seqNo == SequenceNumbers.UNASSIGNED_SEQ_NO : "old op recovering but it already has a seq no.;" +
" index version: " + engineConfig.getIndexSettings().getIndexVersionCreated() + ", seqNo: " + seqNo;
} else if (origin == Operation.Origin.PRIMARY) {
// sequence number should not be set when operation origin is primary
assert seqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO : "primary ops should never have an assigned seq no.; seqNo: " + seqNo;
assert seqNo == SequenceNumbers.UNASSIGNED_SEQ_NO : "primary ops should never have an assigned seq no.; seqNo: " + seqNo;
} else if (engineConfig.getIndexSettings().getIndexVersionCreated().onOrAfter(Version.V_6_0_0_alpha1)) {
// sequence number should be set when operation origin is not primary
assert seqNo >= 0 : "recovery or replica ops should have an assigned seq no.; origin: " + origin;
@ -651,7 +651,7 @@ public class InternalEngine extends Engine {
final Translog.Location location;
if (indexResult.hasFailure() == false) {
location = translog.add(new Translog.Index(index, indexResult));
} else if (indexResult.getSeqNo() != SequenceNumbersService.UNASSIGNED_SEQ_NO) {
} else if (indexResult.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
// if we have document failure, record it as a no-op in the translog with the generated seq_no
location = translog.add(new Translog.NoOp(indexResult.getSeqNo(), index.primaryTerm(), indexResult.getFailure().getMessage()));
} else {
@ -659,7 +659,7 @@ public class InternalEngine extends Engine {
}
indexResult.setTranslogLocation(location);
}
if (indexResult.getSeqNo() != SequenceNumbersService.UNASSIGNED_SEQ_NO) {
if (indexResult.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
seqNoService().markSeqNoAsCompleted(indexResult.getSeqNo());
}
indexResult.setTook(System.nanoTime() - index.startTime());
@ -692,7 +692,7 @@ public class InternalEngine extends Engine {
// this allows to ignore the case where a document was found in the live version maps in
// a delete state and return false for the created flag in favor of code simplicity
final OpVsLuceneDocStatus opVsLucene;
if (index.seqNo() == SequenceNumbersService.UNASSIGNED_SEQ_NO) {
if (index.seqNo() == SequenceNumbers.UNASSIGNED_SEQ_NO) {
// This can happen if the primary is still on an old node and send traffic without seq# or we recover from translog
// created by an old version.
assert config().getIndexSettings().getIndexVersionCreated().before(Version.V_6_0_0_alpha1) :
@ -873,7 +873,7 @@ public class InternalEngine extends Engine {
VersionConflictEngineException e, boolean currentNotFoundOrDeleted, long currentVersion) {
final IndexResult result = new IndexResult(e, currentVersion);
return new IndexingStrategy(
currentNotFoundOrDeleted, false, false, SequenceNumbersService.UNASSIGNED_SEQ_NO, Versions.NOT_FOUND, result);
currentNotFoundOrDeleted, false, false, SequenceNumbers.UNASSIGNED_SEQ_NO, Versions.NOT_FOUND, result);
}
static IndexingStrategy processNormally(boolean currentNotFoundOrDeleted,
@ -951,7 +951,7 @@ public class InternalEngine extends Engine {
final Translog.Location location;
if (deleteResult.hasFailure() == false) {
location = translog.add(new Translog.Delete(delete, deleteResult));
} else if (deleteResult.getSeqNo() != SequenceNumbersService.UNASSIGNED_SEQ_NO) {
} else if (deleteResult.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
location = translog.add(new Translog.NoOp(deleteResult.getSeqNo(),
delete.primaryTerm(), deleteResult.getFailure().getMessage()));
} else {
@ -959,7 +959,7 @@ public class InternalEngine extends Engine {
}
deleteResult.setTranslogLocation(location);
}
if (deleteResult.getSeqNo() != SequenceNumbersService.UNASSIGNED_SEQ_NO) {
if (deleteResult.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
seqNoService().markSeqNoAsCompleted(deleteResult.getSeqNo());
}
deleteResult.setTook(System.nanoTime() - delete.startTime());
@ -987,7 +987,7 @@ public class InternalEngine extends Engine {
// this allows to ignore the case where a document was found in the live version maps in
// a delete state and return true for the found flag in favor of code simplicity
final OpVsLuceneDocStatus opVsLucene;
if (delete.seqNo() == SequenceNumbersService.UNASSIGNED_SEQ_NO) {
if (delete.seqNo() == SequenceNumbers.UNASSIGNED_SEQ_NO) {
assert config().getIndexSettings().getIndexVersionCreated().before(Version.V_6_0_0_alpha1) :
"index is newly created but op has no sequence numbers. op: " + delete;
opVsLucene = compareOpToLuceneDocBasedOnVersions(delete);
@ -1091,7 +1091,7 @@ public class InternalEngine extends Engine {
static DeletionStrategy skipDueToVersionConflict(
VersionConflictEngineException e, long currentVersion, boolean currentlyDeleted) {
final long unassignedSeqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO;
final long unassignedSeqNo = SequenceNumbers.UNASSIGNED_SEQ_NO;
final DeleteResult deleteResult = new DeleteResult(e, currentVersion, unassignedSeqNo, currentlyDeleted == false);
return new DeletionStrategy(false, currentlyDeleted, unassignedSeqNo, Versions.NOT_FOUND, deleteResult);
}
@ -1127,7 +1127,7 @@ public class InternalEngine extends Engine {
private NoOpResult innerNoOp(final NoOp noOp) throws IOException {
assert readLock.isHeldByCurrentThread() || writeLock.isHeldByCurrentThread();
assert noOp.seqNo() > SequenceNumbersService.NO_OPS_PERFORMED;
assert noOp.seqNo() > SequenceNumbers.NO_OPS_PERFORMED;
final long seqNo = noOp.seqNo();
try {
final NoOpResult noOpResult = new NoOpResult(noOp.seqNo());
@ -1137,7 +1137,7 @@ public class InternalEngine extends Engine {
noOpResult.freeze();
return noOpResult;
} finally {
if (seqNo != SequenceNumbersService.UNASSIGNED_SEQ_NO) {
if (seqNo != SequenceNumbers.UNASSIGNED_SEQ_NO) {
seqNoService().markSeqNoAsCompleted(seqNo);
}
}

6
core/src/main/java/org/elasticsearch/index/mapper/SeqNoFieldMapper.java
View File

@ -35,7 +35,7 @@ import org.elasticsearch.index.fielddata.IndexNumericFieldData.NumericType;
import org.elasticsearch.index.fielddata.plain.DocValuesIndexFieldData;
import org.elasticsearch.index.mapper.ParseContext.Document;
import org.elasticsearch.index.query.QueryShardContext;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import java.io.IOException;
import java.util.List;
@ -78,8 +78,8 @@ public class SeqNoFieldMapper extends MetadataFieldMapper {
}
public static SequenceIDFields emptySeqID() {
return new SequenceIDFields(new LongPoint(NAME, SequenceNumbersService.UNASSIGNED_SEQ_NO),
new NumericDocValuesField(NAME, SequenceNumbersService.UNASSIGNED_SEQ_NO),
return new SequenceIDFields(new LongPoint(NAME, SequenceNumbers.UNASSIGNED_SEQ_NO),
new NumericDocValuesField(NAME, SequenceNumbers.UNASSIGNED_SEQ_NO),
new NumericDocValuesField(PRIMARY_TERM_NAME, 0));
}
}

28
core/src/main/java/org/elasticsearch/index/seqno/GlobalCheckpointTracker.java
View File

@ -195,7 +195,7 @@ public class GlobalCheckpointTracker extends AbstractIndexShardComponent {
private boolean invariant() {
// local checkpoints only set during primary mode
assert primaryMode || localCheckpoints.values().stream()
.allMatch(lcps -> lcps.localCheckpoint == SequenceNumbersService.UNASSIGNED_SEQ_NO ||
.allMatch(lcps -> lcps.localCheckpoint == SequenceNumbers.UNASSIGNED_SEQ_NO ||
lcps.localCheckpoint == SequenceNumbersService.PRE_60_NODE_LOCAL_CHECKPOINT);
// relocation handoff can only occur in primary mode
@ -242,15 +242,15 @@ public class GlobalCheckpointTracker extends AbstractIndexShardComponent {
/**
* Initialize the global checkpoint service. The specified global checkpoint should be set to the last known global checkpoint, or
* {@link SequenceNumbersService#UNASSIGNED_SEQ_NO}.
* {@link SequenceNumbers#UNASSIGNED_SEQ_NO}.
*
* @param shardId the shard ID
* @param indexSettings the index settings
* @param globalCheckpoint the last known global checkpoint for this shard, or {@link SequenceNumbersService#UNASSIGNED_SEQ_NO}
* @param globalCheckpoint the last known global checkpoint for this shard, or {@link SequenceNumbers#UNASSIGNED_SEQ_NO}
*/
GlobalCheckpointTracker(final ShardId shardId, final IndexSettings indexSettings, final long globalCheckpoint) {
super(shardId, indexSettings);
assert globalCheckpoint >= SequenceNumbersService.UNASSIGNED_SEQ_NO : "illegal initial global checkpoint: " + globalCheckpoint;
assert globalCheckpoint >= SequenceNumbers.UNASSIGNED_SEQ_NO : "illegal initial global checkpoint: " + globalCheckpoint;
this.primaryMode = false;
this.handoffInProgress = false;
this.appliedClusterStateVersion = -1L;
@ -314,9 +314,9 @@ public class GlobalCheckpointTracker extends AbstractIndexShardComponent {
assert invariant();
assert primaryMode == false;
assert localCheckpoints.get(allocationId) != null && localCheckpoints.get(allocationId).inSync &&
localCheckpoints.get(allocationId).localCheckpoint == SequenceNumbersService.UNASSIGNED_SEQ_NO :
localCheckpoints.get(allocationId).localCheckpoint == SequenceNumbers.UNASSIGNED_SEQ_NO :
"expected " + allocationId + " to have initialized entry in " + localCheckpoints + " when activating primary";
assert localCheckpoint >= SequenceNumbersService.NO_OPS_PERFORMED;
assert localCheckpoint >= SequenceNumbers.NO_OPS_PERFORMED;
primaryMode = true;
updateLocalCheckpoint(allocationId, localCheckpoints.get(allocationId), localCheckpoint);
updateGlobalCheckpointOnPrimary();
@ -354,19 +354,19 @@ public class GlobalCheckpointTracker extends AbstractIndexShardComponent {
assert inSync == false : "update from master in primary mode has " + initializingId +
" as in-sync but it does not exist locally";
final long localCheckpoint = pre60AllocationIds.contains(initializingId) ?
SequenceNumbersService.PRE_60_NODE_LOCAL_CHECKPOINT : SequenceNumbersService.UNASSIGNED_SEQ_NO;
SequenceNumbersService.PRE_60_NODE_LOCAL_CHECKPOINT : SequenceNumbers.UNASSIGNED_SEQ_NO;
localCheckpoints.put(initializingId, new LocalCheckpointState(localCheckpoint, inSync));
}
}
} else {
for (String initializingId : initializingAllocationIds) {
final long localCheckpoint = pre60AllocationIds.contains(initializingId) ?
SequenceNumbersService.PRE_60_NODE_LOCAL_CHECKPOINT : SequenceNumbersService.UNASSIGNED_SEQ_NO;
SequenceNumbersService.PRE_60_NODE_LOCAL_CHECKPOINT : SequenceNumbers.UNASSIGNED_SEQ_NO;
localCheckpoints.put(initializingId, new LocalCheckpointState(localCheckpoint, false));
}
for (String inSyncId : inSyncAllocationIds) {
final long localCheckpoint = pre60AllocationIds.contains(inSyncId) ?
SequenceNumbersService.PRE_60_NODE_LOCAL_CHECKPOINT : SequenceNumbersService.UNASSIGNED_SEQ_NO;
SequenceNumbersService.PRE_60_NODE_LOCAL_CHECKPOINT : SequenceNumbers.UNASSIGNED_SEQ_NO;
localCheckpoints.put(inSyncId, new LocalCheckpointState(localCheckpoint, true));
}
}
@ -413,7 +413,7 @@ public class GlobalCheckpointTracker extends AbstractIndexShardComponent {
// can happen if replica was removed from cluster but recovery process is unaware of it yet
throw new IllegalStateException("no local checkpoint tracking information available for " + allocationId);
}
assert localCheckpoint >= SequenceNumbersService.NO_OPS_PERFORMED :
assert localCheckpoint >= SequenceNumbers.NO_OPS_PERFORMED :
"expected known local checkpoint for " + allocationId + " but was " + localCheckpoint;
assert pendingInSync.contains(allocationId) == false : "shard copy " + allocationId + " is already marked as pending in-sync";
updateLocalCheckpoint(allocationId, lcps, localCheckpoint);
@ -451,7 +451,7 @@ public class GlobalCheckpointTracker extends AbstractIndexShardComponent {
localCheckpoint == SequenceNumbersService.PRE_60_NODE_LOCAL_CHECKPOINT :
"pre-6.0 shard copy " + allocationId + " unexpected to send valid local checkpoint " + localCheckpoint;
// a local checkpoint for a shard copy should be a valid sequence number or the pre-6.0 sequence number indicator
assert localCheckpoint != SequenceNumbersService.UNASSIGNED_SEQ_NO :
assert localCheckpoint != SequenceNumbers.UNASSIGNED_SEQ_NO :
"invalid local checkpoint for shard copy [" + allocationId + "]";
if (localCheckpoint > lcps.localCheckpoint) {
logger.trace("updated local checkpoint of [{}] from [{}] to [{}]", allocationId, lcps.localCheckpoint, localCheckpoint);
@ -508,7 +508,7 @@ public class GlobalCheckpointTracker extends AbstractIndexShardComponent {
}
for (final LocalCheckpointState lcps : localCheckpoints) {
if (lcps.inSync) {
if (lcps.localCheckpoint == SequenceNumbersService.UNASSIGNED_SEQ_NO) {
if (lcps.localCheckpoint == SequenceNumbers.UNASSIGNED_SEQ_NO) {
// unassigned in-sync replica
return fallback;
} else if (lcps.localCheckpoint == SequenceNumbersService.PRE_60_NODE_LOCAL_CHECKPOINT) {
@ -580,9 +580,9 @@ public class GlobalCheckpointTracker extends AbstractIndexShardComponent {
handoffInProgress = false;
// forget all checkpoint information
localCheckpoints.values().stream().forEach(lcps -> {
if (lcps.localCheckpoint != SequenceNumbersService.UNASSIGNED_SEQ_NO &&
if (lcps.localCheckpoint != SequenceNumbers.UNASSIGNED_SEQ_NO &&
lcps.localCheckpoint != SequenceNumbersService.PRE_60_NODE_LOCAL_CHECKPOINT) {
lcps.localCheckpoint = SequenceNumbersService.UNASSIGNED_SEQ_NO;
lcps.localCheckpoint = SequenceNumbers.UNASSIGNED_SEQ_NO;
}
});
assert invariant();

22
core/src/main/java/org/elasticsearch/index/seqno/LocalCheckpointTracker.java
View File

@ -67,26 +67,26 @@ public class LocalCheckpointTracker {
/**
* Initialize the local checkpoint service. The {@code maxSeqNo} should be set to the last sequence number assigned, or
* {@link SequenceNumbersService#NO_OPS_PERFORMED} and {@code localCheckpoint} should be set to the last known local checkpoint,
* or {@link SequenceNumbersService#NO_OPS_PERFORMED}.
* {@link SequenceNumbers#NO_OPS_PERFORMED} and {@code localCheckpoint} should be set to the last known local checkpoint,
* or {@link SequenceNumbers#NO_OPS_PERFORMED}.
*
* @param indexSettings the index settings
* @param maxSeqNo the last sequence number assigned, or {@link SequenceNumbersService#NO_OPS_PERFORMED}
* @param localCheckpoint the last known local checkpoint, or {@link SequenceNumbersService#NO_OPS_PERFORMED}
* @param maxSeqNo the last sequence number assigned, or {@link SequenceNumbers#NO_OPS_PERFORMED}
* @param localCheckpoint the last known local checkpoint, or {@link SequenceNumbers#NO_OPS_PERFORMED}
*/
public LocalCheckpointTracker(final IndexSettings indexSettings, final long maxSeqNo, final long localCheckpoint) {
if (localCheckpoint < 0 && localCheckpoint != SequenceNumbersService.NO_OPS_PERFORMED) {
if (localCheckpoint < 0 && localCheckpoint != SequenceNumbers.NO_OPS_PERFORMED) {
throw new IllegalArgumentException(
"local checkpoint must be non-negative or [" + SequenceNumbersService.NO_OPS_PERFORMED + "] "
"local checkpoint must be non-negative or [" + SequenceNumbers.NO_OPS_PERFORMED + "] "
+ "but was [" + localCheckpoint + "]");
}
if (maxSeqNo < 0 && maxSeqNo != SequenceNumbersService.NO_OPS_PERFORMED) {
if (maxSeqNo < 0 && maxSeqNo != SequenceNumbers.NO_OPS_PERFORMED) {
throw new IllegalArgumentException(
"max seq. no. must be non-negative or [" + SequenceNumbersService.NO_OPS_PERFORMED + "] but was [" + maxSeqNo + "]");
"max seq. no. must be non-negative or [" + SequenceNumbers.NO_OPS_PERFORMED + "] but was [" + maxSeqNo + "]");
}
bitArraysSize = SETTINGS_BIT_ARRAYS_SIZE.get(indexSettings.getSettings());
firstProcessedSeqNo = localCheckpoint == SequenceNumbersService.NO_OPS_PERFORMED ? 0 : localCheckpoint + 1;
nextSeqNo = maxSeqNo == SequenceNumbersService.NO_OPS_PERFORMED ? 0 : maxSeqNo + 1;
firstProcessedSeqNo = localCheckpoint == SequenceNumbers.NO_OPS_PERFORMED ? 0 : localCheckpoint + 1;
nextSeqNo = maxSeqNo == SequenceNumbers.NO_OPS_PERFORMED ? 0 : maxSeqNo + 1;
checkpoint = localCheckpoint;
}
@ -127,7 +127,7 @@ public class LocalCheckpointTracker {
* @param checkpoint the local checkpoint to reset this tracker to
*/
synchronized void resetCheckpoint(final long checkpoint) {
assert checkpoint != SequenceNumbersService.UNASSIGNED_SEQ_NO;
assert checkpoint != SequenceNumbers.UNASSIGNED_SEQ_NO;
assert checkpoint <= this.checkpoint;
processedSeqNo.clear();
firstProcessedSeqNo = checkpoint + 1;

44
core/src/main/java/org/elasticsearch/index/seqno/SequenceNumbers.java
View File

@ -28,6 +28,14 @@ public class SequenceNumbers {
public static final String LOCAL_CHECKPOINT_KEY = "local_checkpoint";
public static final String MAX_SEQ_NO = "max_seq_no";
/**
* Represents an unassigned sequence number (e.g., can be used on primary operations before they are executed).
*/
public static final long UNASSIGNED_SEQ_NO = -2L;
/**
* Represents no operations have been performed on the shard.
*/
public static final long NO_OPS_PERFORMED = -1L;
/**
* Reads the sequence number stats from the commit data (maximum sequence number and local checkpoint) and uses the specified global
@ -40,16 +48,16 @@ public class SequenceNumbers {
public static SeqNoStats loadSeqNoStatsFromLuceneCommit(
final long globalCheckpoint,
final Iterable<Map.Entry<String, String>> commitData) {
long maxSeqNo = SequenceNumbersService.NO_OPS_PERFORMED;
long localCheckpoint = SequenceNumbersService.NO_OPS_PERFORMED;
long maxSeqNo = NO_OPS_PERFORMED;
long localCheckpoint = NO_OPS_PERFORMED;
for (final Map.Entry<String, String> entry : commitData) {
final String key = entry.getKey();
if (key.equals(SequenceNumbers.LOCAL_CHECKPOINT_KEY)) {
assert localCheckpoint == SequenceNumbersService.NO_OPS_PERFORMED : localCheckpoint;
assert localCheckpoint == NO_OPS_PERFORMED : localCheckpoint;
localCheckpoint = Long.parseLong(entry.getValue());
} else if (key.equals(SequenceNumbers.MAX_SEQ_NO)) {
assert maxSeqNo == SequenceNumbersService.NO_OPS_PERFORMED : maxSeqNo;
assert maxSeqNo == NO_OPS_PERFORMED : maxSeqNo;
maxSeqNo = Long.parseLong(entry.getValue());
}
}
@ -59,22 +67,22 @@ public class SequenceNumbers {
/**
* Compute the minimum of the given current minimum sequence number and the specified sequence number, accounting for the fact that the
* current minimum sequence number could be {@link SequenceNumbersService#NO_OPS_PERFORMED} or
* {@link SequenceNumbersService#UNASSIGNED_SEQ_NO}. When the current minimum sequence number is not
* {@link SequenceNumbersService#NO_OPS_PERFORMED} nor {@link SequenceNumbersService#UNASSIGNED_SEQ_NO}, the specified sequence number
* must not be {@link SequenceNumbersService#UNASSIGNED_SEQ_NO}.
* current minimum sequence number could be {@link SequenceNumbers#NO_OPS_PERFORMED} or
* {@link SequenceNumbers#UNASSIGNED_SEQ_NO}. When the current minimum sequence number is not
* {@link SequenceNumbers#NO_OPS_PERFORMED} nor {@link SequenceNumbers#UNASSIGNED_SEQ_NO}, the specified sequence number
* must not be {@link SequenceNumbers#UNASSIGNED_SEQ_NO}.
*
* @param minSeqNo the current minimum sequence number
* @param seqNo the specified sequence number
* @return the new minimum sequence number
*/
public static long min(final long minSeqNo, final long seqNo) {
if (minSeqNo == SequenceNumbersService.NO_OPS_PERFORMED) {
if (minSeqNo == NO_OPS_PERFORMED) {
return seqNo;
} else if (minSeqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO) {
} else if (minSeqNo == UNASSIGNED_SEQ_NO) {
return seqNo;
} else {
if (seqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO) {
if (seqNo == UNASSIGNED_SEQ_NO) {
throw new IllegalArgumentException("sequence number must be assigned");
}
return Math.min(minSeqNo, seqNo);
@ -83,22 +91,22 @@ public class SequenceNumbers {
/**
* Compute the maximum of the given current maximum sequence number and the specified sequence number, accounting for the fact that the
* current maximum sequence number could be {@link SequenceNumbersService#NO_OPS_PERFORMED} or
* {@link SequenceNumbersService#UNASSIGNED_SEQ_NO}. When the current maximum sequence number is not
* {@link SequenceNumbersService#NO_OPS_PERFORMED} nor {@link SequenceNumbersService#UNASSIGNED_SEQ_NO}, the specified sequence number
* must not be {@link SequenceNumbersService#UNASSIGNED_SEQ_NO}.
* current maximum sequence number could be {@link SequenceNumbers#NO_OPS_PERFORMED} or
* {@link SequenceNumbers#UNASSIGNED_SEQ_NO}. When the current maximum sequence number is not
* {@link SequenceNumbers#NO_OPS_PERFORMED} nor {@link SequenceNumbers#UNASSIGNED_SEQ_NO}, the specified sequence number
* must not be {@link SequenceNumbers#UNASSIGNED_SEQ_NO}.
*
* @param maxSeqNo the current maximum sequence number
* @param seqNo the specified sequence number
* @return the new maximum sequence number
*/
public static long max(final long maxSeqNo, final long seqNo) {
if (maxSeqNo == SequenceNumbersService.NO_OPS_PERFORMED) {
if (maxSeqNo == NO_OPS_PERFORMED) {
return seqNo;
} else if (maxSeqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO) {
} else if (maxSeqNo == UNASSIGNED_SEQ_NO) {
return seqNo;
} else {
if (seqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO) {
if (seqNo == UNASSIGNED_SEQ_NO) {
throw new IllegalArgumentException("sequence number must be assigned");
}
return Math.max(maxSeqNo, seqNo);

22
core/src/main/java/org/elasticsearch/index/seqno/SequenceNumbersService.java
View File

@ -32,16 +32,6 @@ import java.util.Set;
*/
public class SequenceNumbersService extends AbstractIndexShardComponent {
/**
* Represents an unassigned sequence number (e.g., can be used on primary operations before they are executed).
*/
public static final long UNASSIGNED_SEQ_NO = -2L;
/**
* Represents no operations have been performed on the shard.
*/
public static final long NO_OPS_PERFORMED = -1L;
/**
* Represents a local checkpoint coming from a pre-6.0 node
*/
@ -52,15 +42,15 @@ public class SequenceNumbersService extends AbstractIndexShardComponent {
/**
* Initialize the sequence number service. The {@code maxSeqNo} should be set to the last sequence number assigned by this shard, or
* {@link SequenceNumbersService#NO_OPS_PERFORMED}, {@code localCheckpoint} should be set to the last known local checkpoint for this
* shard, or {@link SequenceNumbersService#NO_OPS_PERFORMED}, and {@code globalCheckpoint} should be set to the last known global
* checkpoint for this shard, or {@link SequenceNumbersService#UNASSIGNED_SEQ_NO}.
* {@link SequenceNumbers#NO_OPS_PERFORMED}, {@code localCheckpoint} should be set to the last known local checkpoint for this
* shard, or {@link SequenceNumbers#NO_OPS_PERFORMED}, and {@code globalCheckpoint} should be set to the last known global
* checkpoint for this shard, or {@link SequenceNumbers#UNASSIGNED_SEQ_NO}.
*
* @param shardId the shard this service is providing tracking local checkpoints for
* @param indexSettings the index settings
* @param maxSeqNo the last sequence number assigned by this shard, or {@link SequenceNumbersService#NO_OPS_PERFORMED}
* @param localCheckpoint the last known local checkpoint for this shard, or {@link SequenceNumbersService#NO_OPS_PERFORMED}
* @param globalCheckpoint the last known global checkpoint for this shard, or {@link SequenceNumbersService#UNASSIGNED_SEQ_NO}
* @param maxSeqNo the last sequence number assigned by this shard, or {@link SequenceNumbers#NO_OPS_PERFORMED}
* @param localCheckpoint the last known local checkpoint for this shard, or {@link SequenceNumbers#NO_OPS_PERFORMED}
* @param globalCheckpoint the last known global checkpoint for this shard, or {@link SequenceNumbers#UNASSIGNED_SEQ_NO}
*/
public SequenceNumbersService(
final ShardId shardId,

9
core/src/main/java/org/elasticsearch/index/shard/IndexShard.java
View File

@ -104,6 +104,7 @@ import org.elasticsearch.index.search.stats.SearchStats;
import org.elasticsearch.index.search.stats.ShardSearchStats;
import org.elasticsearch.index.seqno.GlobalCheckpointTracker;
import org.elasticsearch.index.seqno.SeqNoStats;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.shard.PrimaryReplicaSyncer.ResyncTask;
import org.elasticsearch.index.similarity.SimilarityService;
@ -631,7 +632,7 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
public Engine.IndexResult applyIndexOperationOnPrimary(long version, VersionType versionType, SourceToParse sourceToParse,
long autoGeneratedTimestamp, boolean isRetry,
Consumer<Mapping> onMappingUpdate) throws IOException {
return applyIndexOperation(SequenceNumbersService.UNASSIGNED_SEQ_NO, primaryTerm, version, versionType, autoGeneratedTimestamp,
return applyIndexOperation(SequenceNumbers.UNASSIGNED_SEQ_NO, primaryTerm, version, versionType, autoGeneratedTimestamp,
isRetry, Engine.Operation.Origin.PRIMARY, sourceToParse, onMappingUpdate);
}
@ -728,7 +729,7 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
public Engine.DeleteResult applyDeleteOperationOnPrimary(long version, String type, String id, VersionType versionType,
Consumer<Mapping> onMappingUpdate) throws IOException {
return applyDeleteOperation(SequenceNumbersService.UNASSIGNED_SEQ_NO, primaryTerm, version, type, id, versionType,
return applyDeleteOperation(SequenceNumbers.UNASSIGNED_SEQ_NO, primaryTerm, version, type, id, versionType,
Engine.Operation.Origin.PRIMARY, onMappingUpdate);
}
@ -2134,8 +2135,8 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
updateGlobalCheckpointOnReplica(globalCheckpoint, "primary term transition");
final long currentGlobalCheckpoint = getGlobalCheckpoint();
final long localCheckpoint;
if (currentGlobalCheckpoint == SequenceNumbersService.UNASSIGNED_SEQ_NO) {
localCheckpoint = SequenceNumbersService.NO_OPS_PERFORMED;
if (currentGlobalCheckpoint == SequenceNumbers.UNASSIGNED_SEQ_NO) {
localCheckpoint = SequenceNumbers.NO_OPS_PERFORMED;
} else {
localCheckpoint = currentGlobalCheckpoint;
}

4
core/src/main/java/org/elasticsearch/index/shard/PrimaryReplicaSyncer.java
View File

@ -35,7 +35,7 @@ import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.util.concurrent.AbstractRunnable;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.translog.Translog;
import org.elasticsearch.tasks.Task;
import org.elasticsearch.tasks.TaskId;
@ -231,7 +231,7 @@ public class PrimaryReplicaSyncer extends AbstractComponent {
while ((operation = snapshot.next()) != null) {
final long seqNo = operation.seqNo();
if (startingSeqNo >= 0 &&
(seqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO || seqNo < startingSeqNo)) {
(seqNo == SequenceNumbers.UNASSIGNED_SEQ_NO || seqNo < startingSeqNo)) {
totalSkippedOps.incrementAndGet();
continue;
}

12
core/src/main/java/org/elasticsearch/index/translog/Checkpoint.java
View File

@ -28,7 +28,7 @@ import org.apache.lucene.store.IndexInput;
import org.apache.lucene.store.OutputStreamIndexOutput;
import org.apache.lucene.store.SimpleFSDirectory;
import org.elasticsearch.common.io.Channels;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
@ -105,8 +105,8 @@ final class Checkpoint {
static Checkpoint emptyTranslogCheckpoint(final long offset, final long generation, final long globalCheckpoint,
long minTranslogGeneration) {
final long minSeqNo = SequenceNumbersService.NO_OPS_PERFORMED;
final long maxSeqNo = SequenceNumbersService.NO_OPS_PERFORMED;
final long minSeqNo = SequenceNumbers.NO_OPS_PERFORMED;
final long maxSeqNo = SequenceNumbers.NO_OPS_PERFORMED;
return new Checkpoint(offset, 0, generation, minSeqNo, maxSeqNo, globalCheckpoint, minTranslogGeneration);
}
@ -116,9 +116,9 @@ final class Checkpoint {
// reads a checksummed checkpoint introduced in ES 5.0.0
static Checkpoint readCheckpointV5_0_0(final DataInput in) throws IOException {
final long minSeqNo = SequenceNumbersService.NO_OPS_PERFORMED;
final long maxSeqNo = SequenceNumbersService.NO_OPS_PERFORMED;
final long globalCheckpoint = SequenceNumbersService.UNASSIGNED_SEQ_NO;
final long minSeqNo = SequenceNumbers.NO_OPS_PERFORMED;
final long maxSeqNo = SequenceNumbers.NO_OPS_PERFORMED;
final long globalCheckpoint = SequenceNumbers.UNASSIGNED_SEQ_NO;
final long minTranslogGeneration = -1L;
return new Checkpoint(in.readLong(), in.readInt(), in.readLong(), minSeqNo, maxSeqNo, globalCheckpoint, minTranslogGeneration);
}

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

@ -43,7 +43,7 @@ import org.elasticsearch.index.IndexSettings;
import org.elasticsearch.index.VersionType;
import org.elasticsearch.index.engine.Engine;
import org.elasticsearch.index.mapper.Uid;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.shard.AbstractIndexShardComponent;
import org.elasticsearch.index.shard.IndexShardComponent;
@ -634,7 +634,7 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC
return Stream.concat(readers.stream(), Stream.of(current))
.filter(reader -> {
final long maxSeqNo = reader.getCheckpoint().maxSeqNo;
return maxSeqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO || maxSeqNo >= minSeqNo;
return maxSeqNo == SequenceNumbers.UNASSIGNED_SEQ_NO || maxSeqNo >= minSeqNo;
});
}
@ -978,7 +978,7 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC
seqNo = in.readLong();
primaryTerm = in.readLong();
} else {
seqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO;
seqNo = SequenceNumbers.UNASSIGNED_SEQ_NO;
primaryTerm = 0;
}
}
@ -1182,7 +1182,7 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC
seqNo = in.readLong();
primaryTerm = in.readLong();
} else {
seqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO;
seqNo = SequenceNumbers.UNASSIGNED_SEQ_NO;
primaryTerm = 0;
}
}
@ -1329,7 +1329,7 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC
}
public NoOp(final long seqNo, final long primaryTerm, final String reason) {
assert seqNo > SequenceNumbersService.NO_OPS_PERFORMED;
assert seqNo > SequenceNumbers.NO_OPS_PERFORMED;
assert primaryTerm >= 0;
assert reason != null;
this.seqNo = seqNo;

11
core/src/main/java/org/elasticsearch/index/translog/TranslogWriter.java
View File

@ -31,7 +31,6 @@ import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.io.Channels;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.shard.ShardId;
import java.io.BufferedOutputStream;
@ -97,9 +96,9 @@ public class TranslogWriter extends BaseTranslogReader implements Closeable {
this.outputStream = new BufferedChannelOutputStream(java.nio.channels.Channels.newOutputStream(channel), bufferSize.bytesAsInt());
this.lastSyncedCheckpoint = initialCheckpoint;
this.totalOffset = initialCheckpoint.offset;
assert initialCheckpoint.minSeqNo == SequenceNumbersService.NO_OPS_PERFORMED : initialCheckpoint.minSeqNo;
assert initialCheckpoint.minSeqNo == SequenceNumbers.NO_OPS_PERFORMED : initialCheckpoint.minSeqNo;
this.minSeqNo = initialCheckpoint.minSeqNo;
assert initialCheckpoint.maxSeqNo == SequenceNumbersService.NO_OPS_PERFORMED : initialCheckpoint.maxSeqNo;
assert initialCheckpoint.maxSeqNo == SequenceNumbers.NO_OPS_PERFORMED : initialCheckpoint.maxSeqNo;
this.maxSeqNo = initialCheckpoint.maxSeqNo;
this.globalCheckpointSupplier = globalCheckpointSupplier;
this.seenSequenceNumbers = Assertions.ENABLED ? new HashMap<>() : null;
@ -193,10 +192,10 @@ public class TranslogWriter extends BaseTranslogReader implements Closeable {
}
totalOffset += data.length();
if (minSeqNo == SequenceNumbersService.NO_OPS_PERFORMED) {
if (minSeqNo == SequenceNumbers.NO_OPS_PERFORMED) {
assert operationCounter == 0;
}
if (maxSeqNo == SequenceNumbersService.NO_OPS_PERFORMED) {
if (maxSeqNo == SequenceNumbers.NO_OPS_PERFORMED) {
assert operationCounter == 0;
}
@ -211,7 +210,7 @@ public class TranslogWriter extends BaseTranslogReader implements Closeable {
}
private synchronized boolean assertNoSeqNumberConflict(long seqNo, BytesReference data) throws IOException {
if (seqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO) {
if (seqNo == SequenceNumbers.UNASSIGNED_SEQ_NO) {
// nothing to do
} else if (seenSequenceNumbers.containsKey(seqNo)) {
final Tuple<BytesReference, Exception> previous = seenSequenceNumbers.get(seqNo);

4
core/src/main/java/org/elasticsearch/index/translog/TruncateTranslogCommand.java
View File

@ -40,7 +40,7 @@ import org.elasticsearch.common.SuppressForbidden;
import org.elasticsearch.common.io.PathUtils;
import org.elasticsearch.env.Environment;
import org.elasticsearch.index.IndexNotFoundException;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import java.io.IOException;
import java.nio.channels.Channels;
@ -169,7 +169,7 @@ public class TruncateTranslogCommand extends EnvironmentAwareCommand {
/** Write a checkpoint file to the given location with the given generation */
public static void writeEmptyCheckpoint(Path filename, int translogLength, long translogGeneration) throws IOException {
Checkpoint emptyCheckpoint = Checkpoint.emptyTranslogCheckpoint(translogLength, translogGeneration,
SequenceNumbersService.UNASSIGNED_SEQ_NO, translogGeneration);
SequenceNumbers.UNASSIGNED_SEQ_NO, translogGeneration);
Checkpoint.write(FileChannel::open, filename, emptyCheckpoint,
StandardOpenOption.WRITE, StandardOpenOption.READ, StandardOpenOption.CREATE_NEW);
// fsync with metadata here to make sure.

12
core/src/main/java/org/elasticsearch/indices/recovery/PeerRecoveryTargetService.java
View File

@ -42,7 +42,7 @@ import org.elasticsearch.index.IndexNotFoundException;
import org.elasticsearch.index.engine.RecoveryEngineException;
import org.elasticsearch.index.mapper.MapperException;
import org.elasticsearch.index.seqno.SeqNoStats;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.shard.IllegalIndexShardStateException;
import org.elasticsearch.index.shard.IndexEventListener;
import org.elasticsearch.index.shard.IndexShard;
@ -319,10 +319,10 @@ public class PeerRecoveryTargetService extends AbstractComponent implements Inde
if (metadataSnapshot.size() > 0) {
startingSeqNo = getStartingSeqNo(recoveryTarget);
} else {
startingSeqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO;
startingSeqNo = SequenceNumbers.UNASSIGNED_SEQ_NO;
}
if (startingSeqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO) {
if (startingSeqNo == SequenceNumbers.UNASSIGNED_SEQ_NO) {
logger.trace("{} preparing for file-based recovery from [{}]", recoveryTarget.shardId(), recoveryTarget.sourceNode());
} else {
logger.trace(
@ -348,7 +348,7 @@ public class PeerRecoveryTargetService extends AbstractComponent implements Inde
* Get the starting sequence number for a sequence-number-based request.
*
* @param recoveryTarget the target of the recovery
* @return the starting sequence number or {@link SequenceNumbersService#UNASSIGNED_SEQ_NO} if obtaining the starting sequence number
* @return the starting sequence number or {@link SequenceNumbers#UNASSIGNED_SEQ_NO} if obtaining the starting sequence number
* failed
*/
public static long getStartingSeqNo(final RecoveryTarget recoveryTarget) {
@ -364,7 +364,7 @@ public class PeerRecoveryTargetService extends AbstractComponent implements Inde
*/
return seqNoStats.getLocalCheckpoint() + 1;
} else {
return SequenceNumbersService.UNASSIGNED_SEQ_NO;
return SequenceNumbers.UNASSIGNED_SEQ_NO;
}
} catch (final IOException e) {
/*
@ -372,7 +372,7 @@ public class PeerRecoveryTargetService extends AbstractComponent implements Inde
* translog on the recovery target is opened, the recovery enters a retry loop seeing now that the index files are on disk and
* proceeds to attempt a sequence-number-based recovery.
*/
return SequenceNumbersService.UNASSIGNED_SEQ_NO;
return SequenceNumbers.UNASSIGNED_SEQ_NO;
}
}

4
core/src/main/java/org/elasticsearch/indices/recovery/RecoveryFinalizeRecoveryRequest.java
View File

@ -22,7 +22,7 @@ package org.elasticsearch.indices.recovery;
import org.elasticsearch.Version;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.transport.TransportRequest;
@ -63,7 +63,7 @@ public class RecoveryFinalizeRecoveryRequest extends TransportRequest {
if (in.getVersion().onOrAfter(Version.V_6_0_0_alpha1)) {
globalCheckpoint = in.readZLong();
} else {
globalCheckpoint = SequenceNumbersService.UNASSIGNED_SEQ_NO;
globalCheckpoint = SequenceNumbers.UNASSIGNED_SEQ_NO;
}
}

14
core/src/main/java/org/elasticsearch/indices/recovery/RecoverySourceHandler.java
View File

@ -47,7 +47,7 @@ import org.elasticsearch.common.util.CancellableThreads;
import org.elasticsearch.index.engine.Engine;
import org.elasticsearch.index.engine.RecoveryEngineException;
import org.elasticsearch.index.seqno.LocalCheckpointTracker;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.shard.IndexShard;
import org.elasticsearch.index.shard.IndexShardClosedException;
import org.elasticsearch.index.shard.IndexShardRelocatedException;
@ -147,7 +147,7 @@ public class RecoverySourceHandler {
final Translog translog = shard.getTranslog();
final long startingSeqNo;
boolean isSequenceNumberBasedRecoveryPossible = request.startingSeqNo() != SequenceNumbersService.UNASSIGNED_SEQ_NO &&
boolean isSequenceNumberBasedRecoveryPossible = request.startingSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO &&
isTranslogReadyForSequenceNumberBasedRecovery();
if (isSequenceNumberBasedRecoveryPossible) {
@ -162,7 +162,7 @@ public class RecoverySourceHandler {
}
// we set this to unassigned to create a translog roughly according to the retention policy
// on the target
startingSeqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO;
startingSeqNo = SequenceNumbers.UNASSIGNED_SEQ_NO;
try {
phase1(phase1Snapshot.getIndexCommit(), translog::totalOperations);
@ -239,7 +239,7 @@ public class RecoverySourceHandler {
try (Translog.Snapshot snapshot = shard.getTranslog().newSnapshotFromMinSeqNo(startingSeqNo)) {
Translog.Operation operation;
while ((operation = snapshot.next()) != null) {
if (operation.seqNo() != SequenceNumbersService.UNASSIGNED_SEQ_NO) {
if (operation.seqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
tracker.markSeqNoAsCompleted(operation.seqNo());
}
}
@ -427,7 +427,7 @@ public class RecoverySourceHandler {
* point-in-time view of the translog). It then sends each translog operation to the target node so it can be replayed into the new
* shard.
*
* @param startingSeqNo the sequence number to start recovery from, or {@link SequenceNumbersService#UNASSIGNED_SEQ_NO} if all
* @param startingSeqNo the sequence number to start recovery from, or {@link SequenceNumbers#UNASSIGNED_SEQ_NO} if all
* ops should be sent
* @param snapshot a snapshot of the translog
*
@ -513,7 +513,7 @@ public class RecoverySourceHandler {
long size = 0;
int skippedOps = 0;
int totalSentOps = 0;
final AtomicLong targetLocalCheckpoint = new AtomicLong(SequenceNumbersService.UNASSIGNED_SEQ_NO);
final AtomicLong targetLocalCheckpoint = new AtomicLong(SequenceNumbers.UNASSIGNED_SEQ_NO);
final List<Translog.Operation> operations = new ArrayList<>();
final int expectedTotalOps = snapshot.totalOperations();
@ -536,7 +536,7 @@ public class RecoverySourceHandler {
* any ops before the starting sequence number.
*/
final long seqNo = operation.seqNo();
if (startingSeqNo >= 0 && (seqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO || seqNo < startingSeqNo)) {
if (startingSeqNo >= 0 && (seqNo == SequenceNumbers.UNASSIGNED_SEQ_NO || seqNo < startingSeqNo)) {
skippedOps++;
continue;
}

4
core/src/main/java/org/elasticsearch/indices/recovery/RecoveryTranslogOperationsResponse.java
View File

@ -22,7 +22,7 @@ package org.elasticsearch.indices.recovery;
import org.elasticsearch.Version;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.transport.FutureTransportResponseHandler;
import org.elasticsearch.transport.TransportResponse;
import org.elasticsearch.transport.TransportResponseHandler;
@ -56,7 +56,7 @@ public class RecoveryTranslogOperationsResponse extends TransportResponse {
localCheckpoint = in.readZLong();
}
else {
localCheckpoint = SequenceNumbersService.UNASSIGNED_SEQ_NO;
localCheckpoint = SequenceNumbers.UNASSIGNED_SEQ_NO;
}
}

4
core/src/main/java/org/elasticsearch/indices/recovery/StartRecoveryRequest.java
View File

@ -23,7 +23,7 @@ import org.elasticsearch.Version;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.index.store.Store;
import org.elasticsearch.transport.TransportRequest;
@ -122,7 +122,7 @@ public class StartRecoveryRequest extends TransportRequest {
if (in.getVersion().onOrAfter(Version.V_6_0_0_alpha1)) {
startingSeqNo = in.readLong();
} else {
startingSeqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO;
startingSeqNo = SequenceNumbers.UNASSIGNED_SEQ_NO;
}
}

10
core/src/test/java/org/elasticsearch/action/DocWriteResponseTests.java
View File

@ -25,7 +25,7 @@ import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.json.JsonXContent;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.test.ESTestCase;
@ -42,7 +42,7 @@ public class DocWriteResponseTests extends ESTestCase {
new ShardId("index", "uuid", 0),
"type",
"id",
SequenceNumbersService.UNASSIGNED_SEQ_NO,
SequenceNumbers.UNASSIGNED_SEQ_NO,
17,
0,
Result.CREATED) {};
@ -56,7 +56,7 @@ public class DocWriteResponseTests extends ESTestCase {
new ShardId("index", "uuid", 0),
"type",
"❤",
SequenceNumbersService.UNASSIGNED_SEQ_NO,
SequenceNumbers.UNASSIGNED_SEQ_NO,
17,
0,
Result.CREATED) {};
@ -70,7 +70,7 @@ public class DocWriteResponseTests extends ESTestCase {
new ShardId("index", "uuid", 0),
"type",
"a b",
SequenceNumbersService.UNASSIGNED_SEQ_NO,
SequenceNumbers.UNASSIGNED_SEQ_NO,
17,
0,
Result.CREATED) {};
@ -88,7 +88,7 @@ public class DocWriteResponseTests extends ESTestCase {
new ShardId("index", "uuid", 0),
"type",
"id",
SequenceNumbersService.UNASSIGNED_SEQ_NO,
SequenceNumbers.UNASSIGNED_SEQ_NO,
17,
0,
Result.CREATED) {

6
core/src/test/java/org/elasticsearch/action/delete/DeleteResponseTests.java
View File

@ -26,7 +26,7 @@ import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.test.ESTestCase;
import org.elasticsearch.test.RandomObjects;
@ -114,8 +114,8 @@ public class DeleteResponseTests extends ESTestCase {
int shardId = randomIntBetween(0, 5);
String type = randomAlphaOfLength(5);
String id = randomAlphaOfLength(5);
long seqNo = randomFrom(SequenceNumbersService.UNASSIGNED_SEQ_NO, randomNonNegativeLong(), (long) randomIntBetween(0, 10000));
long primaryTerm = seqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO ? 0 : randomIntBetween(1, 10000);
long seqNo = randomFrom(SequenceNumbers.UNASSIGNED_SEQ_NO, randomNonNegativeLong(), (long) randomIntBetween(0, 10000));
long primaryTerm = seqNo == SequenceNumbers.UNASSIGNED_SEQ_NO ? 0 : randomIntBetween(1, 10000);
long version = randomBoolean() ? randomNonNegativeLong() : randomIntBetween(0, 10000);
boolean found = randomBoolean();
boolean forcedRefresh = randomBoolean();

9
core/src/test/java/org/elasticsearch/action/index/IndexRequestTests.java
View File

@ -26,10 +26,10 @@ import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.common.bytes.BytesArray;
import org.elasticsearch.common.io.stream.BytesStreamOutput;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.VersionType;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.test.ESTestCase;
@ -37,7 +37,6 @@ import org.elasticsearch.test.ESTestCase;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.Arrays;
import java.util.Base64;
import java.util.HashSet;
import java.util.Set;
@ -135,7 +134,7 @@ public class IndexRequestTests extends ESTestCase {
String id = randomAlphaOfLengthBetween(3, 10);
long version = randomLong();
boolean created = randomBoolean();
IndexResponse indexResponse = new IndexResponse(shardId, type, id, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, version, created);
IndexResponse indexResponse = new IndexResponse(shardId, type, id, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, version, created);
int total = randomIntBetween(1, 10);
int successful = randomIntBetween(1, 10);
ReplicationResponse.ShardInfo shardInfo = new ReplicationResponse.ShardInfo(total, successful);
@ -155,7 +154,7 @@ public class IndexRequestTests extends ESTestCase {
assertEquals(forcedRefresh, indexResponse.forcedRefresh());
assertEquals("IndexResponse[index=" + shardId.getIndexName() + ",type=" + type + ",id="+ id +
",version=" + version + ",result=" + (created ? "created" : "updated") +
",seqNo=" + SequenceNumbersService.UNASSIGNED_SEQ_NO +
",seqNo=" + SequenceNumbers.UNASSIGNED_SEQ_NO +
",primaryTerm=" + 0 +
",shards={\"total\":" + total + ",\"successful\":" + successful + ",\"failed\":0}]",
indexResponse.toString());

6
core/src/test/java/org/elasticsearch/action/index/IndexResponseTests.java
View File

@ -27,7 +27,7 @@ import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.xcontent.ToXContent;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.test.ESTestCase;
import org.elasticsearch.test.RandomObjects;
@ -127,8 +127,8 @@ public class IndexResponseTests extends ESTestCase {
int shardId = randomIntBetween(0, 5);
String type = randomAlphaOfLength(5);
String id = randomAlphaOfLength(5);
long seqNo = randomFrom(SequenceNumbersService.UNASSIGNED_SEQ_NO, randomNonNegativeLong(), (long) randomIntBetween(0, 10000));
long primaryTerm = seqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO ? 0 : randomIntBetween(1, 10000);
long seqNo = randomFrom(SequenceNumbers.UNASSIGNED_SEQ_NO, randomNonNegativeLong(), (long) randomIntBetween(0, 10000));
long primaryTerm = seqNo == SequenceNumbers.UNASSIGNED_SEQ_NO ? 0 : randomIntBetween(1, 10000);
long version = randomBoolean() ? randomNonNegativeLong() : randomIntBetween(0, 10000);
boolean created = randomBoolean();
boolean forcedRefresh = randomBoolean();

134
core/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java
View File

@ -346,7 +346,7 @@ public class InternalEngineTests extends ESTestCase {
protected Translog createTranslog(Path translogPath) throws IOException {
TranslogConfig translogConfig = new TranslogConfig(shardId, translogPath, INDEX_SETTINGS, BigArrays.NON_RECYCLING_INSTANCE);
return new Translog(translogConfig, null, createTranslogDeletionPolicy(INDEX_SETTINGS), () -> SequenceNumbersService.UNASSIGNED_SEQ_NO);
return new Translog(translogConfig, null, createTranslogDeletionPolicy(INDEX_SETTINGS), () -> SequenceNumbers.UNASSIGNED_SEQ_NO);
}
protected InternalEngine createEngine(Store store, Path translogPath) throws IOException {
@ -721,9 +721,9 @@ public class InternalEngineTests extends ESTestCase {
}
public void testCommitStats() throws IOException {
final AtomicLong maxSeqNo = new AtomicLong(SequenceNumbersService.NO_OPS_PERFORMED);
final AtomicLong localCheckpoint = new AtomicLong(SequenceNumbersService.NO_OPS_PERFORMED);
final AtomicLong globalCheckpoint = new AtomicLong(SequenceNumbersService.UNASSIGNED_SEQ_NO);
final AtomicLong maxSeqNo = new AtomicLong(SequenceNumbers.NO_OPS_PERFORMED);
final AtomicLong localCheckpoint = new AtomicLong(SequenceNumbers.NO_OPS_PERFORMED);
final AtomicLong globalCheckpoint = new AtomicLong(SequenceNumbers.UNASSIGNED_SEQ_NO);
try (
Store store = createStore();
InternalEngine engine = createEngine(store, createTempDir(), (config) -> new SequenceNumbersService(
@ -740,19 +740,19 @@ public class InternalEngineTests extends ESTestCase {
assertThat(stats1.getUserData(), hasKey(SequenceNumbers.LOCAL_CHECKPOINT_KEY));
assertThat(
Long.parseLong(stats1.getUserData().get(SequenceNumbers.LOCAL_CHECKPOINT_KEY)),
equalTo(SequenceNumbersService.NO_OPS_PERFORMED));
equalTo(SequenceNumbers.NO_OPS_PERFORMED));
assertThat(stats1.getUserData(), hasKey(SequenceNumbers.MAX_SEQ_NO));
assertThat(
Long.parseLong(stats1.getUserData().get(SequenceNumbers.MAX_SEQ_NO)),
equalTo(SequenceNumbersService.NO_OPS_PERFORMED));
equalTo(SequenceNumbers.NO_OPS_PERFORMED));
maxSeqNo.set(rarely() ? SequenceNumbersService.NO_OPS_PERFORMED : randomIntBetween(0, 1024));
maxSeqNo.set(rarely() ? SequenceNumbers.NO_OPS_PERFORMED : randomIntBetween(0, 1024));
localCheckpoint.set(
rarely() || maxSeqNo.get() == SequenceNumbersService.NO_OPS_PERFORMED ?
SequenceNumbersService.NO_OPS_PERFORMED : randomIntBetween(0, 1024));
globalCheckpoint.set(rarely() || localCheckpoint.get() == SequenceNumbersService.NO_OPS_PERFORMED ?
SequenceNumbersService.UNASSIGNED_SEQ_NO : randomIntBetween(0, (int) localCheckpoint.get()));
rarely() || maxSeqNo.get() == SequenceNumbers.NO_OPS_PERFORMED ?
SequenceNumbers.NO_OPS_PERFORMED : randomIntBetween(0, 1024));
globalCheckpoint.set(rarely() || localCheckpoint.get() == SequenceNumbers.NO_OPS_PERFORMED ?
SequenceNumbers.UNASSIGNED_SEQ_NO : randomIntBetween(0, (int) localCheckpoint.get()));
engine.flush(true, true);
@ -827,11 +827,11 @@ public class InternalEngineTests extends ESTestCase {
for (int i = 0; i < ops; i++) {
final ParsedDocument doc = testParsedDocument("1", null, testDocumentWithTextField(), SOURCE, null);
if (randomBoolean()) {
final Engine.Index operation = new Engine.Index(newUid(doc), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, i, VersionType.EXTERNAL, Engine.Operation.Origin.PRIMARY, System.nanoTime(), -1, false);
final Engine.Index operation = new Engine.Index(newUid(doc), doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, i, VersionType.EXTERNAL, Engine.Operation.Origin.PRIMARY, System.nanoTime(), -1, false);
operations.add(operation);
initialEngine.index(operation);
} else {
final Engine.Delete operation = new Engine.Delete("test", "1", newUid(doc), SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, i, VersionType.EXTERNAL, Engine.Operation.Origin.PRIMARY, System.nanoTime());
final Engine.Delete operation = new Engine.Delete("test", "1", newUid(doc), SequenceNumbers.UNASSIGNED_SEQ_NO, 0, i, VersionType.EXTERNAL, Engine.Operation.Origin.PRIMARY, System.nanoTime());
operations.add(operation);
initialEngine.delete(operation);
}
@ -902,9 +902,9 @@ public class InternalEngineTests extends ESTestCase {
new SequenceNumbersService(
config.getShardId(),
config.getIndexSettings(),
SequenceNumbersService.NO_OPS_PERFORMED,
SequenceNumbersService.NO_OPS_PERFORMED,
SequenceNumbersService.UNASSIGNED_SEQ_NO) {
SequenceNumbers.NO_OPS_PERFORMED,
SequenceNumbers.NO_OPS_PERFORMED,
SequenceNumbers.UNASSIGNED_SEQ_NO) {
@Override
public long generateSeqNo() {
return seqNos.get(counter.getAndIncrement());
@ -1215,7 +1215,7 @@ public class InternalEngineTests extends ESTestCase {
final boolean forceMergeFlushes = randomBoolean();
final ParsedDocument parsedDoc3 = testParsedDocument("3", null, testDocumentWithTextField(), B_1, null);
if (forceMergeFlushes) {
engine.index(new Engine.Index(newUid(parsedDoc3), parsedDoc3, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_ANY, VersionType.INTERNAL, Engine.Operation.Origin.PRIMARY, System.nanoTime() - engine.engineConfig.getFlushMergesAfter().nanos(), -1, false));
engine.index(new Engine.Index(newUid(parsedDoc3), parsedDoc3, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_ANY, VersionType.INTERNAL, Engine.Operation.Origin.PRIMARY, System.nanoTime() - engine.engineConfig.getFlushMergesAfter().nanos(), -1, false));
} else {
engine.index(indexForDoc(parsedDoc3));
}
@ -1417,11 +1417,11 @@ public class InternalEngineTests extends ESTestCase {
public void testVersioningCreateExistsException() throws IOException {
ParsedDocument doc = testParsedDocument("1", null, testDocument(), B_1, null);
Engine.Index create = new Engine.Index(newUid(doc), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_DELETED, VersionType.INTERNAL, PRIMARY, 0, -1, false);
Engine.Index create = new Engine.Index(newUid(doc), doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_DELETED, VersionType.INTERNAL, PRIMARY, 0, -1, false);
Engine.IndexResult indexResult = engine.index(create);
assertThat(indexResult.getVersion(), equalTo(1L));
create = new Engine.Index(newUid(doc), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_DELETED, VersionType.INTERNAL, PRIMARY, 0, -1, false);
create = new Engine.Index(newUid(doc), doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_DELETED, VersionType.INTERNAL, PRIMARY, 0, -1, false);
indexResult = engine.index(create);
assertTrue(indexResult.hasFailure());
assertThat(indexResult.getFailure(), instanceOf(VersionConflictEngineException.class));
@ -1462,7 +1462,7 @@ public class InternalEngineTests extends ESTestCase {
}
if (randomBoolean()) {
op = new Engine.Index(id, testParsedDocument("1", null, testDocumentWithTextField(valuePrefix + i), B_1, null),
forReplica && i >= startWithSeqNo ? i * 2 : SequenceNumbersService.UNASSIGNED_SEQ_NO,
forReplica && i >= startWithSeqNo ? i * 2 : SequenceNumbers.UNASSIGNED_SEQ_NO,
forReplica && i >= startWithSeqNo && incrementTermWhenIntroducingSeqNo ? primaryTerm + 1 : primaryTerm,
version,
forReplica ? versionType.versionTypeForReplicationAndRecovery() : versionType,
@ -1471,7 +1471,7 @@ public class InternalEngineTests extends ESTestCase {
);
} else {
op = new Engine.Delete("test", "1", id,
forReplica && i >= startWithSeqNo ? i * 2 : SequenceNumbersService.UNASSIGNED_SEQ_NO,
forReplica && i >= startWithSeqNo ? i * 2 : SequenceNumbers.UNASSIGNED_SEQ_NO,
forReplica && i >= startWithSeqNo && incrementTermWhenIntroducingSeqNo ? primaryTerm + 1 : primaryTerm,
version,
forReplica ? versionType.versionTypeForReplicationAndRecovery() : versionType,
@ -1900,7 +1900,7 @@ public class InternalEngineTests extends ESTestCase {
Engine.Index index = new Engine.Index(uidTerm,
testParsedDocument("1", null, testDocument(),
bytesArray(Strings.collectionToCommaDelimitedString(values)), null),
SequenceNumbersService.UNASSIGNED_SEQ_NO, 2,
SequenceNumbers.UNASSIGNED_SEQ_NO, 2,
get.version(), VersionType.INTERNAL,
PRIMARY, System.currentTimeMillis(), -1, false);
Engine.IndexResult indexResult = engine.index(index);
@ -2017,13 +2017,13 @@ public class InternalEngineTests extends ESTestCase {
public void testSeqNoAndCheckpoints() throws IOException {
final int opCount = randomIntBetween(1, 256);
long primarySeqNo = SequenceNumbersService.NO_OPS_PERFORMED;
long primarySeqNo = SequenceNumbers.NO_OPS_PERFORMED;
final String[] ids = new String[]{"1", "2", "3"};
final Set<String> indexedIds = new HashSet<>();
long localCheckpoint = SequenceNumbersService.NO_OPS_PERFORMED;
long replicaLocalCheckpoint = SequenceNumbersService.NO_OPS_PERFORMED;
long globalCheckpoint = SequenceNumbersService.UNASSIGNED_SEQ_NO;
long maxSeqNo = SequenceNumbersService.NO_OPS_PERFORMED;
long localCheckpoint = SequenceNumbers.NO_OPS_PERFORMED;
long replicaLocalCheckpoint = SequenceNumbers.NO_OPS_PERFORMED;
long globalCheckpoint = SequenceNumbers.UNASSIGNED_SEQ_NO;
long maxSeqNo = SequenceNumbers.NO_OPS_PERFORMED;
InternalEngine initialEngine = null;
try {
@ -2041,7 +2041,7 @@ public class InternalEngineTests extends ESTestCase {
// we have some docs indexed, so delete one of them
id = randomFrom(indexedIds);
final Engine.Delete delete = new Engine.Delete(
"test", id, newUid(id), SequenceNumbersService.UNASSIGNED_SEQ_NO, 0,
"test", id, newUid(id), SequenceNumbers.UNASSIGNED_SEQ_NO, 0,
rarely() ? 100 : Versions.MATCH_ANY, VersionType.INTERNAL, PRIMARY, 0);
final Engine.DeleteResult result = initialEngine.delete(delete);
if (!result.hasFailure()) {
@ -2050,7 +2050,7 @@ public class InternalEngineTests extends ESTestCase {
indexedIds.remove(id);
primarySeqNo++;
} else {
assertThat(result.getSeqNo(), equalTo(SequenceNumbersService.UNASSIGNED_SEQ_NO));
assertThat(result.getSeqNo(), equalTo(SequenceNumbers.UNASSIGNED_SEQ_NO));
assertThat(initialEngine.seqNoService().getMaxSeqNo(), equalTo(primarySeqNo));
}
} else {
@ -2058,7 +2058,7 @@ public class InternalEngineTests extends ESTestCase {
id = randomFrom(ids);
ParsedDocument doc = testParsedDocument(id, null, testDocumentWithTextField(), SOURCE, null);
final Engine.Index index = new Engine.Index(newUid(doc), doc,
SequenceNumbersService.UNASSIGNED_SEQ_NO, 0,
SequenceNumbers.UNASSIGNED_SEQ_NO, 0,
rarely() ? 100 : Versions.MATCH_ANY, VersionType.INTERNAL,
PRIMARY, 0, -1, false);
final Engine.IndexResult result = initialEngine.index(index);
@ -2068,7 +2068,7 @@ public class InternalEngineTests extends ESTestCase {
indexedIds.add(id);
primarySeqNo++;
} else {
assertThat(result.getSeqNo(), equalTo(SequenceNumbersService.UNASSIGNED_SEQ_NO));
assertThat(result.getSeqNo(), equalTo(SequenceNumbers.UNASSIGNED_SEQ_NO));
assertThat(initialEngine.seqNoService().getMaxSeqNo(), equalTo(primarySeqNo));
}
}
@ -2192,17 +2192,17 @@ public class InternalEngineTests extends ESTestCase {
} while (doneIndexing == false);
// now, verify all the commits have the correct docs according to the user commit data
long prevLocalCheckpoint = SequenceNumbersService.NO_OPS_PERFORMED;
long prevMaxSeqNo = SequenceNumbersService.NO_OPS_PERFORMED;
long prevLocalCheckpoint = SequenceNumbers.NO_OPS_PERFORMED;
long prevMaxSeqNo = SequenceNumbers.NO_OPS_PERFORMED;
for (Engine.IndexCommitRef commitRef : commits) {
final IndexCommit commit = commitRef.getIndexCommit();
Map<String, String> userData = commit.getUserData();
long localCheckpoint = userData.containsKey(SequenceNumbers.LOCAL_CHECKPOINT_KEY) ?
Long.parseLong(userData.get(SequenceNumbers.LOCAL_CHECKPOINT_KEY)) :
SequenceNumbersService.NO_OPS_PERFORMED;
SequenceNumbers.NO_OPS_PERFORMED;
long maxSeqNo = userData.containsKey(SequenceNumbers.MAX_SEQ_NO) ?
Long.parseLong(userData.get(SequenceNumbers.MAX_SEQ_NO)) :
SequenceNumbersService.UNASSIGNED_SEQ_NO;
SequenceNumbers.UNASSIGNED_SEQ_NO;
// local checkpoint and max seq no shouldn't go backwards
assertThat(localCheckpoint, greaterThanOrEqualTo(prevLocalCheckpoint));
assertThat(maxSeqNo, greaterThanOrEqualTo(prevMaxSeqNo));
@ -2212,7 +2212,7 @@ public class InternalEngineTests extends ESTestCase {
if (highest != null) {
highestSeqNo = highest.longValue();
} else {
highestSeqNo = SequenceNumbersService.NO_OPS_PERFORMED;
highestSeqNo = SequenceNumbers.NO_OPS_PERFORMED;
}
// make sure localCheckpoint <= highest seq no found <= maxSeqNo
assertThat(highestSeqNo, greaterThanOrEqualTo(localCheckpoint));
@ -2317,10 +2317,10 @@ public class InternalEngineTests extends ESTestCase {
document.add(new TextField("value", "test1", Field.Store.YES));
ParsedDocument doc = testParsedDocument("1", null, document, B_2, null);
engine.index(new Engine.Index(newUid(doc), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, 1, VersionType.EXTERNAL, Engine.Operation.Origin.PRIMARY, System.nanoTime(), -1, false));
engine.index(new Engine.Index(newUid(doc), doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, 1, VersionType.EXTERNAL, Engine.Operation.Origin.PRIMARY, System.nanoTime(), -1, false));
// Delete document we just added:
engine.delete(new Engine.Delete("test", "1", newUid(doc), SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, 10, VersionType.EXTERNAL, Engine.Operation.Origin.PRIMARY, System.nanoTime()));
engine.delete(new Engine.Delete("test", "1", newUid(doc), SequenceNumbers.UNASSIGNED_SEQ_NO, 0, 10, VersionType.EXTERNAL, Engine.Operation.Origin.PRIMARY, System.nanoTime()));
// Get should not find the document
Engine.GetResult getResult = engine.get(newGet(true, doc), searcherFactory);
@ -2334,14 +2334,14 @@ public class InternalEngineTests extends ESTestCase {
}
// Delete non-existent document
engine.delete(new Engine.Delete("test", "2", newUid("2"), SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, 10, VersionType.EXTERNAL, Engine.Operation.Origin.PRIMARY, System.nanoTime()));
engine.delete(new Engine.Delete("test", "2", newUid("2"), SequenceNumbers.UNASSIGNED_SEQ_NO, 0, 10, VersionType.EXTERNAL, Engine.Operation.Origin.PRIMARY, System.nanoTime()));
// Get should not find the document (we never indexed uid=2):
getResult = engine.get(new Engine.Get(true, "type", "2", newUid("2")), searcherFactory);
assertThat(getResult.exists(), equalTo(false));
// Try to index uid=1 with a too-old version, should fail:
Engine.Index index = new Engine.Index(newUid(doc), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, 2, VersionType.EXTERNAL, Engine.Operation.Origin.PRIMARY, System.nanoTime(), -1, false);
Engine.Index index = new Engine.Index(newUid(doc), doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, 2, VersionType.EXTERNAL, Engine.Operation.Origin.PRIMARY, System.nanoTime(), -1, false);
Engine.IndexResult indexResult = engine.index(index);
assertTrue(indexResult.hasFailure());
assertThat(indexResult.getFailure(), instanceOf(VersionConflictEngineException.class));
@ -2351,7 +2351,7 @@ public class InternalEngineTests extends ESTestCase {
assertThat(getResult.exists(), equalTo(false));
// Try to index uid=2 with a too-old version, should fail:
Engine.Index index1 = new Engine.Index(newUid(doc), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, 2, VersionType.EXTERNAL, Engine.Operation.Origin.PRIMARY, System.nanoTime(), -1, false);
Engine.Index index1 = new Engine.Index(newUid(doc), doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, 2, VersionType.EXTERNAL, Engine.Operation.Origin.PRIMARY, System.nanoTime(), -1, false);
indexResult = engine.index(index1);
assertTrue(indexResult.hasFailure());
assertThat(indexResult.getFailure(), instanceOf(VersionConflictEngineException.class));
@ -2469,7 +2469,7 @@ public class InternalEngineTests extends ESTestCase {
final int numDocs = randomIntBetween(1, 10);
for (int i = 0; i < numDocs; i++) {
ParsedDocument doc = testParsedDocument(Integer.toString(i), null, testDocument(), new BytesArray("{}"), null);
Engine.Index firstIndexRequest = new Engine.Index(newUid(doc), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_DELETED, VersionType.INTERNAL, PRIMARY, System.nanoTime(), -1, false);
Engine.Index firstIndexRequest = new Engine.Index(newUid(doc), doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_DELETED, VersionType.INTERNAL, PRIMARY, System.nanoTime(), -1, false);
Engine.IndexResult indexResult = engine.index(firstIndexRequest);
assertThat(indexResult.getVersion(), equalTo(1L));
}
@ -2568,7 +2568,7 @@ public class InternalEngineTests extends ESTestCase {
final int numDocs = randomIntBetween(1, 10);
for (int i = 0; i < numDocs; i++) {
ParsedDocument doc = testParsedDocument(Integer.toString(i), null, testDocument(), new BytesArray("{}"), null);
Engine.Index firstIndexRequest = new Engine.Index(newUid(doc), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_DELETED, VersionType.INTERNAL, PRIMARY, System.nanoTime(), -1, false);
Engine.Index firstIndexRequest = new Engine.Index(newUid(doc), doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_DELETED, VersionType.INTERNAL, PRIMARY, System.nanoTime(), -1, false);
Engine.IndexResult indexResult = engine.index(firstIndexRequest);
assertThat(indexResult.getVersion(), equalTo(1L));
}
@ -2603,7 +2603,7 @@ public class InternalEngineTests extends ESTestCase {
final int numDocs = randomIntBetween(1, 10);
for (int i = 0; i < numDocs; i++) {
ParsedDocument doc = testParsedDocument(Integer.toString(i), null, testDocument(), new BytesArray("{}"), null);
Engine.Index firstIndexRequest = new Engine.Index(newUid(doc), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_DELETED, VersionType.INTERNAL, PRIMARY, System.nanoTime(), -1, false);
Engine.Index firstIndexRequest = new Engine.Index(newUid(doc), doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_DELETED, VersionType.INTERNAL, PRIMARY, System.nanoTime(), -1, false);
Engine.IndexResult indexResult = engine.index(firstIndexRequest);
assertThat(indexResult.getVersion(), equalTo(1L));
}
@ -2635,7 +2635,7 @@ public class InternalEngineTests extends ESTestCase {
final boolean flush = randomBoolean();
int randomId = randomIntBetween(numDocs + 1, numDocs + 10);
ParsedDocument doc = testParsedDocument(Integer.toString(randomId), null, testDocument(), new BytesArray("{}"), null);
Engine.Index firstIndexRequest = new Engine.Index(newUid(doc), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, 1, VersionType.EXTERNAL, PRIMARY, System.nanoTime(), -1, false);
Engine.Index firstIndexRequest = new Engine.Index(newUid(doc), doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, 1, VersionType.EXTERNAL, PRIMARY, System.nanoTime(), -1, false);
Engine.IndexResult indexResult = engine.index(firstIndexRequest);
assertThat(indexResult.getVersion(), equalTo(1L));
if (flush) {
@ -2643,7 +2643,7 @@ public class InternalEngineTests extends ESTestCase {
}
doc = testParsedDocument(Integer.toString(randomId), null, testDocument(), new BytesArray("{}"), null);
Engine.Index idxRequest = new Engine.Index(newUid(doc), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, 2, VersionType.EXTERNAL, PRIMARY, System.nanoTime(), -1, false);
Engine.Index idxRequest = new Engine.Index(newUid(doc), doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, 2, VersionType.EXTERNAL, PRIMARY, System.nanoTime(), -1, false);
Engine.IndexResult result = engine.index(idxRequest);
engine.refresh("test");
assertThat(result.getVersion(), equalTo(2L));
@ -2768,7 +2768,7 @@ public class InternalEngineTests extends ESTestCase {
final int numDocs = randomIntBetween(1, 10);
for (int i = 0; i < numDocs; i++) {
ParsedDocument doc = testParsedDocument(Integer.toString(i), null, testDocument(), new BytesArray("{}"), null);
Engine.Index firstIndexRequest = new Engine.Index(newUid(doc), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_DELETED, VersionType.INTERNAL, PRIMARY, System.nanoTime(), -1, false);
Engine.Index firstIndexRequest = new Engine.Index(newUid(doc), doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_DELETED, VersionType.INTERNAL, PRIMARY, System.nanoTime(), -1, false);
Engine.IndexResult index = engine.index(firstIndexRequest);
assertThat(index.getVersion(), equalTo(1L));
}
@ -2778,7 +2778,7 @@ public class InternalEngineTests extends ESTestCase {
Translog translog = new Translog(
new TranslogConfig(shardId, createTempDir(), INDEX_SETTINGS, BigArrays.NON_RECYCLING_INSTANCE),
null, createTranslogDeletionPolicy(INDEX_SETTINGS), () -> SequenceNumbersService.UNASSIGNED_SEQ_NO);
null, createTranslogDeletionPolicy(INDEX_SETTINGS), () -> SequenceNumbers.UNASSIGNED_SEQ_NO);
translog.add(new Translog.Index("test", "SomeBogusId", 0, "{}".getBytes(Charset.forName("UTF-8"))));
assertEquals(generation.translogFileGeneration, translog.currentFileGeneration());
translog.close();
@ -2904,7 +2904,7 @@ public class InternalEngineTests extends ESTestCase {
// create
{
ParsedDocument doc = testParsedDocument(Integer.toString(0), null, testDocument(), new BytesArray("{}"), null);
Engine.Index firstIndexRequest = new Engine.Index(newUid(doc), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_DELETED, VersionType.INTERNAL, PRIMARY, System.nanoTime(), -1, false);
Engine.Index firstIndexRequest = new Engine.Index(newUid(doc), doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_DELETED, VersionType.INTERNAL, PRIMARY, System.nanoTime(), -1, false);
try (InternalEngine engine = new InternalEngine(copy(config, EngineConfig.OpenMode.CREATE_INDEX_AND_TRANSLOG))){
assertFalse(engine.isRecovering());
@ -3254,7 +3254,7 @@ public class InternalEngineTests extends ESTestCase {
boolean isRetry = false;
long autoGeneratedIdTimestamp = 0;
Engine.Index index = new Engine.Index(newUid(doc), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_ANY, VersionType.INTERNAL, PRIMARY, System.nanoTime(), autoGeneratedIdTimestamp, isRetry);
Engine.Index index = new Engine.Index(newUid(doc), doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_ANY, VersionType.INTERNAL, PRIMARY, System.nanoTime(), autoGeneratedIdTimestamp, isRetry);
Engine.IndexResult indexResult = engine.index(index);
assertThat(indexResult.getVersion(), equalTo(1L));
@ -3263,7 +3263,7 @@ public class InternalEngineTests extends ESTestCase {
assertThat(indexResult.getVersion(), equalTo(1L));
isRetry = true;
index = new Engine.Index(newUid(doc), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_ANY, VersionType.INTERNAL, PRIMARY, System.nanoTime(), autoGeneratedIdTimestamp, isRetry);
index = new Engine.Index(newUid(doc), doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_ANY, VersionType.INTERNAL, PRIMARY, System.nanoTime(), autoGeneratedIdTimestamp, isRetry);
indexResult = engine.index(index);
assertThat(indexResult.getVersion(), equalTo(1L));
engine.refresh("test");
@ -3288,7 +3288,7 @@ public class InternalEngineTests extends ESTestCase {
boolean isRetry = true;
long autoGeneratedIdTimestamp = 0;
Engine.Index firstIndexRequest = new Engine.Index(newUid(doc), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_ANY, VersionType.INTERNAL, PRIMARY, System.nanoTime(), autoGeneratedIdTimestamp, isRetry);
Engine.Index firstIndexRequest = new Engine.Index(newUid(doc), doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_ANY, VersionType.INTERNAL, PRIMARY, System.nanoTime(), autoGeneratedIdTimestamp, isRetry);
Engine.IndexResult result = engine.index(firstIndexRequest);
assertThat(result.getVersion(), equalTo(1L));
@ -3297,7 +3297,7 @@ public class InternalEngineTests extends ESTestCase {
assertThat(indexReplicaResult.getVersion(), equalTo(1L));
isRetry = false;
Engine.Index secondIndexRequest = new Engine.Index(newUid(doc), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_ANY, VersionType.INTERNAL, PRIMARY, System.nanoTime(), autoGeneratedIdTimestamp, isRetry);
Engine.Index secondIndexRequest = new Engine.Index(newUid(doc), doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_ANY, VersionType.INTERNAL, PRIMARY, System.nanoTime(), autoGeneratedIdTimestamp, isRetry);
Engine.IndexResult indexResult = engine.index(secondIndexRequest);
assertTrue(indexResult.isCreated());
engine.refresh("test");
@ -3324,7 +3324,7 @@ public class InternalEngineTests extends ESTestCase {
}
public Engine.Index appendOnlyPrimary(ParsedDocument doc, boolean retry, final long autoGeneratedIdTimestamp) {
return new Engine.Index(newUid(doc), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_ANY,
return new Engine.Index(newUid(doc), doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_ANY,
VersionType.INTERNAL, Engine.Operation.Origin.PRIMARY, System.nanoTime(), autoGeneratedIdTimestamp, retry);
}
@ -3561,7 +3561,7 @@ public class InternalEngineTests extends ESTestCase {
public void testSequenceIDs() throws Exception {
Tuple<Long, Long> seqID = getSequenceID(engine, new Engine.Get(false, "type", "2", newUid("1")));
// Non-existent doc returns no seqnum and no primary term
assertThat(seqID.v1(), equalTo(SequenceNumbersService.UNASSIGNED_SEQ_NO));
assertThat(seqID.v1(), equalTo(SequenceNumbers.UNASSIGNED_SEQ_NO));
assertThat(seqID.v2(), equalTo(0L));
// create a document
@ -3592,7 +3592,7 @@ public class InternalEngineTests extends ESTestCase {
document = testDocumentWithTextField();
document.add(new Field(SourceFieldMapper.NAME, BytesReference.toBytes(B_1), SourceFieldMapper.Defaults.FIELD_TYPE));
doc = testParsedDocument("1", null, document, B_1, null);
engine.index(new Engine.Index(newUid(doc), doc, SequenceNumbersService.UNASSIGNED_SEQ_NO, 3,
engine.index(new Engine.Index(newUid(doc), doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 3,
Versions.MATCH_ANY, VersionType.INTERNAL, Engine.Operation.Origin.PRIMARY,
System.nanoTime(), -1, false));
engine.refresh("test");
@ -3629,9 +3629,9 @@ public class InternalEngineTests extends ESTestCase {
return new SequenceNumbersService(
shardId,
defaultSettings,
SequenceNumbersService.NO_OPS_PERFORMED,
SequenceNumbersService.NO_OPS_PERFORMED,
SequenceNumbersService.UNASSIGNED_SEQ_NO) {
SequenceNumbers.NO_OPS_PERFORMED,
SequenceNumbers.NO_OPS_PERFORMED,
SequenceNumbers.UNASSIGNED_SEQ_NO) {
@Override
public long generateSeqNo() {
final long seqNo = super.generateSeqNo();
@ -3661,7 +3661,7 @@ public class InternalEngineTests extends ESTestCase {
final AtomicReference<CountDownLatch> latchReference = new AtomicReference<>(new CountDownLatch(1));
final CyclicBarrier barrier = new CyclicBarrier(2);
final AtomicBoolean stall = new AtomicBoolean();
final AtomicLong expectedLocalCheckpoint = new AtomicLong(SequenceNumbersService.NO_OPS_PERFORMED);
final AtomicLong expectedLocalCheckpoint = new AtomicLong(SequenceNumbers.NO_OPS_PERFORMED);
final List<Thread> threads = new ArrayList<>();
final SequenceNumbersService seqNoService = getStallingSeqNoService(latchReference, barrier, stall, expectedLocalCheckpoint);
initialEngine = createEngine(defaultSettings, store, primaryTranslogDir, newMergePolicy(), null, (config) -> seqNoService);
@ -3756,7 +3756,7 @@ public class InternalEngineTests extends ESTestCase {
final AtomicLong sequenceNumber = new AtomicLong();
final Engine.Operation.Origin origin = randomFrom(LOCAL_TRANSLOG_RECOVERY, PEER_RECOVERY, PRIMARY, REPLICA);
final LongSupplier sequenceNumberSupplier =
origin == PRIMARY ? () -> SequenceNumbersService.UNASSIGNED_SEQ_NO : sequenceNumber::getAndIncrement;
origin == PRIMARY ? () -> SequenceNumbers.UNASSIGNED_SEQ_NO : sequenceNumber::getAndIncrement;
document.add(new Field(SourceFieldMapper.NAME, BytesReference.toBytes(B_1), SourceFieldMapper.Defaults.FIELD_TYPE));
final ParsedDocument doc = testParsedDocument("1", null, document, B_1, null);
final Term uid = newUid(doc);
@ -3891,7 +3891,7 @@ public class InternalEngineTests extends ESTestCase {
final AtomicReference<CountDownLatch> latchReference = new AtomicReference<>();
final CyclicBarrier barrier = new CyclicBarrier(2);
final AtomicBoolean stall = new AtomicBoolean();
final AtomicLong expectedLocalCheckpoint = new AtomicLong(SequenceNumbersService.NO_OPS_PERFORMED);
final AtomicLong expectedLocalCheckpoint = new AtomicLong(SequenceNumbers.NO_OPS_PERFORMED);
final Map<Thread, CountDownLatch> threads = new LinkedHashMap<>();
final SequenceNumbersService seqNoService = getStallingSeqNoService(latchReference, barrier, stall, expectedLocalCheckpoint);
actualEngine = createEngine(defaultSettings, store, primaryTranslogDir, newMergePolicy(), null, (config) -> seqNoService);
@ -3967,7 +3967,7 @@ public class InternalEngineTests extends ESTestCase {
DocIdAndSeqNo docIdAndSeqNo = VersionsAndSeqNoResolver.loadDocIdAndSeqNo(searcher.reader(), get.uid());
if (docIdAndSeqNo == null) {
primaryTerm = 0;
seqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO;
seqNo = SequenceNumbers.UNASSIGNED_SEQ_NO;
} else {
seqNo = docIdAndSeqNo.seqNo;
primaryTerm = VersionsAndSeqNoResolver.loadPrimaryTerm(docIdAndSeqNo, get.uid().field());
@ -3987,9 +3987,9 @@ public class InternalEngineTests extends ESTestCase {
new SequenceNumbersService(
shardId,
defaultSettings,
SequenceNumbersService.NO_OPS_PERFORMED,
SequenceNumbersService.NO_OPS_PERFORMED,
SequenceNumbersService.UNASSIGNED_SEQ_NO) {
SequenceNumbers.NO_OPS_PERFORMED,
SequenceNumbers.NO_OPS_PERFORMED,
SequenceNumbers.UNASSIGNED_SEQ_NO) {
@Override
public void markSeqNoAsCompleted(long seqNo) {
super.markSeqNoAsCompleted(seqNo);
@ -4025,7 +4025,7 @@ public class InternalEngineTests extends ESTestCase {
}
final long currentLocalCheckpoint = actualEngine.seqNoService().getLocalCheckpoint();
final long resetLocalCheckpoint =
randomIntBetween(Math.toIntExact(SequenceNumbersService.NO_OPS_PERFORMED), Math.toIntExact(currentLocalCheckpoint));
randomIntBetween(Math.toIntExact(SequenceNumbers.NO_OPS_PERFORMED), Math.toIntExact(currentLocalCheckpoint));
actualEngine.seqNoService().resetLocalCheckpoint(resetLocalCheckpoint);
completedSeqNos.clear();
actualEngine.restoreLocalCheckpointFromTranslog();

8
core/src/test/java/org/elasticsearch/index/replication/IndexLevelReplicationTests.java
View File

@ -37,7 +37,7 @@ import org.elasticsearch.index.engine.InternalEngineTests;
import org.elasticsearch.index.engine.SegmentsStats;
import org.elasticsearch.index.engine.VersionConflictEngineException;
import org.elasticsearch.index.seqno.SeqNoStats;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.shard.IndexShard;
import org.elasticsearch.index.shard.IndexShardTests;
import org.elasticsearch.index.store.Store;
@ -165,9 +165,9 @@ public class IndexLevelReplicationTests extends ESIndexLevelReplicationTestCase
*/
final Matcher<Long> globalCheckpointMatcher;
if (shardRouting.primary()) {
globalCheckpointMatcher = numDocs == 0 ? equalTo(SequenceNumbersService.NO_OPS_PERFORMED) : equalTo(numDocs - 1L);
globalCheckpointMatcher = numDocs == 0 ? equalTo(SequenceNumbers.NO_OPS_PERFORMED) : equalTo(numDocs - 1L);
} else {
globalCheckpointMatcher = numDocs == 0 ? equalTo(SequenceNumbersService.NO_OPS_PERFORMED)
globalCheckpointMatcher = numDocs == 0 ? equalTo(SequenceNumbers.NO_OPS_PERFORMED)
: anyOf(equalTo(numDocs - 1L), equalTo(numDocs - 2L));
}
assertThat(shardRouting + " global checkpoint mismatch", shardStats.getGlobalCheckpoint(), globalCheckpointMatcher);
@ -177,7 +177,7 @@ public class IndexLevelReplicationTests extends ESIndexLevelReplicationTestCase
// simulate a background global checkpoint sync at which point we expect the global checkpoint to advance on the replicas
shards.syncGlobalCheckpoint();
final long noOpsPerformed = SequenceNumbersService.NO_OPS_PERFORMED;
final long noOpsPerformed = SequenceNumbers.NO_OPS_PERFORMED;
for (IndexShard shard : shards) {
final SeqNoStats shardStats = shard.seqNoStats();
final ShardRouting shardRouting = shard.routingEntry();

14
core/src/test/java/org/elasticsearch/index/seqno/GlobalCheckpointTrackerTests.java
View File

@ -53,8 +53,8 @@ import java.util.stream.IntStream;
import java.util.stream.Stream;
import static java.util.Collections.emptySet;
import static org.elasticsearch.index.seqno.SequenceNumbersService.NO_OPS_PERFORMED;
import static org.elasticsearch.index.seqno.SequenceNumbersService.UNASSIGNED_SEQ_NO;
import static org.elasticsearch.index.seqno.SequenceNumbers.NO_OPS_PERFORMED;
import static org.elasticsearch.index.seqno.SequenceNumbers.UNASSIGNED_SEQ_NO;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.not;
@ -393,14 +393,14 @@ public class GlobalCheckpointTrackerTests extends ESTestCase {
.stream()
.filter(a -> a.equals(primaryId) == false)
.allMatch(a -> tracker.getTrackedLocalCheckpointForShard(a.getId()).getLocalCheckpoint()
== SequenceNumbersService.UNASSIGNED_SEQ_NO));
== SequenceNumbers.UNASSIGNED_SEQ_NO));
assertTrue(initializingIds.stream().noneMatch(a -> tracker.getTrackedLocalCheckpointForShard(a.getId()).inSync));
assertTrue(
initializingIds
.stream()
.filter(a -> a.equals(primaryId) == false)
.allMatch(a -> tracker.getTrackedLocalCheckpointForShard(a.getId()).getLocalCheckpoint()
== SequenceNumbersService.UNASSIGNED_SEQ_NO));
== SequenceNumbers.UNASSIGNED_SEQ_NO));
// now we will remove some allocation IDs from these and ensure that they propagate through
final Set<AllocationId> removingActiveAllocationIds = new HashSet<>(randomSubsetOf(activeAllocationIds));
@ -433,13 +433,13 @@ public class GlobalCheckpointTrackerTests extends ESTestCase {
.stream()
.filter(a -> a.equals(primaryId) == false)
.allMatch(a -> tracker.getTrackedLocalCheckpointForShard(a.getId()).getLocalCheckpoint()
== SequenceNumbersService.UNASSIGNED_SEQ_NO));
== SequenceNumbers.UNASSIGNED_SEQ_NO));
assertTrue(newInitializingAllocationIds.stream().noneMatch(a -> tracker.getTrackedLocalCheckpointForShard(a.getId()).inSync));
assertTrue(
newInitializingAllocationIds
.stream()
.allMatch(a -> tracker.getTrackedLocalCheckpointForShard(a.getId()).getLocalCheckpoint()
== SequenceNumbersService.UNASSIGNED_SEQ_NO));
== SequenceNumbers.UNASSIGNED_SEQ_NO));
// the tracking allocation IDs should play no role in determining the global checkpoint
final Map<AllocationId, Integer> activeLocalCheckpoints =
@ -737,7 +737,7 @@ public class GlobalCheckpointTrackerTests extends ESTestCase {
private static void randomLocalCheckpointUpdate(GlobalCheckpointTracker gcp) {
String allocationId = randomFrom(gcp.localCheckpoints.keySet());
long currentLocalCheckpoint = gcp.localCheckpoints.get(allocationId).getLocalCheckpoint();
gcp.updateLocalCheckpoint(allocationId, Math.max(SequenceNumbersService.NO_OPS_PERFORMED, currentLocalCheckpoint + randomInt(5)));
gcp.updateLocalCheckpoint(allocationId, Math.max(SequenceNumbers.NO_OPS_PERFORMED, currentLocalCheckpoint + randomInt(5)));
}
private static void randomMarkInSync(GlobalCheckpointTracker gcp) {

12
core/src/test/java/org/elasticsearch/index/seqno/LocalCheckpointTrackerTests.java
View File

@ -56,8 +56,8 @@ public class LocalCheckpointTrackerTests extends ESTestCase {
.builder()
.put(LocalCheckpointTracker.SETTINGS_BIT_ARRAYS_SIZE.getKey(), SMALL_CHUNK_SIZE)
.build()),
SequenceNumbersService.NO_OPS_PERFORMED,
SequenceNumbersService.NO_OPS_PERFORMED
SequenceNumbers.NO_OPS_PERFORMED,
SequenceNumbers.NO_OPS_PERFORMED
);
}
@ -70,7 +70,7 @@ public class LocalCheckpointTrackerTests extends ESTestCase {
public void testSimplePrimary() {
long seqNo1, seqNo2;
assertThat(tracker.getCheckpoint(), equalTo(SequenceNumbersService.NO_OPS_PERFORMED));
assertThat(tracker.getCheckpoint(), equalTo(SequenceNumbers.NO_OPS_PERFORMED));
seqNo1 = tracker.generateSeqNo();
assertThat(seqNo1, equalTo(0L));
tracker.markSeqNoAsCompleted(seqNo1);
@ -86,7 +86,7 @@ public class LocalCheckpointTrackerTests extends ESTestCase {
}
public void testSimpleReplica() {
assertThat(tracker.getCheckpoint(), equalTo(SequenceNumbersService.NO_OPS_PERFORMED));
assertThat(tracker.getCheckpoint(), equalTo(SequenceNumbers.NO_OPS_PERFORMED));
tracker.markSeqNoAsCompleted(0L);
assertThat(tracker.getCheckpoint(), equalTo(0L));
tracker.markSeqNoAsCompleted(2L);
@ -240,7 +240,7 @@ public class LocalCheckpointTrackerTests extends ESTestCase {
public void testResetCheckpoint() {
final int operations = 1024 - scaledRandomIntBetween(0, 1024);
int maxSeqNo = Math.toIntExact(SequenceNumbersService.NO_OPS_PERFORMED);
int maxSeqNo = Math.toIntExact(SequenceNumbers.NO_OPS_PERFORMED);
for (int i = 0; i < operations; i++) {
if (!rarely()) {
tracker.markSeqNoAsCompleted(i);
@ -249,7 +249,7 @@ public class LocalCheckpointTrackerTests extends ESTestCase {
}
final int localCheckpoint =
randomIntBetween(Math.toIntExact(SequenceNumbersService.NO_OPS_PERFORMED), Math.toIntExact(tracker.getCheckpoint()));
randomIntBetween(Math.toIntExact(SequenceNumbers.NO_OPS_PERFORMED), Math.toIntExact(tracker.getCheckpoint()));
tracker.resetCheckpoint(localCheckpoint);
assertThat(tracker.getCheckpoint(), equalTo((long) localCheckpoint));
assertThat(tracker.getMaxSeqNo(), equalTo((long) maxSeqNo));

20
core/src/test/java/org/elasticsearch/index/seqno/SequenceNumbersTests.java
View File

@ -29,31 +29,31 @@ public class SequenceNumbersTests extends ESTestCase {
public void testMin() {
final long seqNo = randomNonNegativeLong();
assertThat(SequenceNumbers.min(SequenceNumbersService.NO_OPS_PERFORMED, seqNo), equalTo(seqNo));
assertThat(SequenceNumbers.min(SequenceNumbers.NO_OPS_PERFORMED, seqNo), equalTo(seqNo));
assertThat(
SequenceNumbers.min(SequenceNumbersService.NO_OPS_PERFORMED, SequenceNumbersService.UNASSIGNED_SEQ_NO),
equalTo(SequenceNumbersService.UNASSIGNED_SEQ_NO));
assertThat(SequenceNumbers.min(SequenceNumbersService.UNASSIGNED_SEQ_NO, seqNo), equalTo(seqNo));
SequenceNumbers.min(SequenceNumbers.NO_OPS_PERFORMED, SequenceNumbers.UNASSIGNED_SEQ_NO),
equalTo(SequenceNumbers.UNASSIGNED_SEQ_NO));
assertThat(SequenceNumbers.min(SequenceNumbers.UNASSIGNED_SEQ_NO, seqNo), equalTo(seqNo));
final long minSeqNo = randomNonNegativeLong();
assertThat(SequenceNumbers.min(minSeqNo, seqNo), equalTo(Math.min(minSeqNo, seqNo)));
final IllegalArgumentException e =
expectThrows(IllegalArgumentException.class, () -> SequenceNumbers.min(minSeqNo, SequenceNumbersService.UNASSIGNED_SEQ_NO));
expectThrows(IllegalArgumentException.class, () -> SequenceNumbers.min(minSeqNo, SequenceNumbers.UNASSIGNED_SEQ_NO));
assertThat(e, hasToString(containsString("sequence number must be assigned")));
}
public void testMax() {
final long seqNo = randomNonNegativeLong();
assertThat(SequenceNumbers.max(SequenceNumbersService.NO_OPS_PERFORMED, seqNo), equalTo(seqNo));
assertThat(SequenceNumbers.max(SequenceNumbers.NO_OPS_PERFORMED, seqNo), equalTo(seqNo));
assertThat(
SequenceNumbers.max(SequenceNumbersService.NO_OPS_PERFORMED, SequenceNumbersService.UNASSIGNED_SEQ_NO),
equalTo(SequenceNumbersService.UNASSIGNED_SEQ_NO));
assertThat(SequenceNumbers.max(SequenceNumbersService.UNASSIGNED_SEQ_NO, seqNo), equalTo(seqNo));
SequenceNumbers.max(SequenceNumbers.NO_OPS_PERFORMED, SequenceNumbers.UNASSIGNED_SEQ_NO),
equalTo(SequenceNumbers.UNASSIGNED_SEQ_NO));
assertThat(SequenceNumbers.max(SequenceNumbers.UNASSIGNED_SEQ_NO, seqNo), equalTo(seqNo));
final long maxSeqNo = randomNonNegativeLong();
assertThat(SequenceNumbers.min(maxSeqNo, seqNo), equalTo(Math.min(maxSeqNo, seqNo)));
final IllegalArgumentException e =
expectThrows(IllegalArgumentException.class, () -> SequenceNumbers.min(maxSeqNo, SequenceNumbersService.UNASSIGNED_SEQ_NO));
expectThrows(IllegalArgumentException.class, () -> SequenceNumbers.min(maxSeqNo, SequenceNumbers.UNASSIGNED_SEQ_NO));
assertThat(e, hasToString(containsString("sequence number must be assigned")));
}

40
core/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java
View File

@ -84,7 +84,7 @@ import org.elasticsearch.index.mapper.ParsedDocument;
import org.elasticsearch.index.mapper.SeqNoFieldMapper;
import org.elasticsearch.index.mapper.SourceToParse;
import org.elasticsearch.index.mapper.Uid;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.snapshots.IndexShardSnapshotStatus;
import org.elasticsearch.index.store.Store;
import org.elasticsearch.index.translog.Translog;
@ -282,7 +282,7 @@ public class IndexShardTests extends IndexShardTestCase {
// expected
}
try {
indexShard.acquireReplicaOperationPermit(indexShard.getPrimaryTerm(), SequenceNumbersService.UNASSIGNED_SEQ_NO, null,
indexShard.acquireReplicaOperationPermit(indexShard.getPrimaryTerm(), SequenceNumbers.UNASSIGNED_SEQ_NO, null,
ThreadPool.Names.INDEX);
fail("we should not be able to increment anymore");
} catch (IndexShardClosedException e) {
@ -294,7 +294,7 @@ public class IndexShardTests extends IndexShardTestCase {
IndexShard indexShard = newShard(false);
expectThrows(IndexShardNotStartedException.class, () ->
indexShard.acquireReplicaOperationPermit(indexShard.getPrimaryTerm() + randomIntBetween(1, 100),
SequenceNumbersService.UNASSIGNED_SEQ_NO, null, ThreadPool.Names.INDEX));
SequenceNumbers.UNASSIGNED_SEQ_NO, null, ThreadPool.Names.INDEX));
closeShards(indexShard);
}
@ -414,7 +414,7 @@ public class IndexShardTests extends IndexShardTestCase {
// most of the time this is large enough that most of the time there will be at least one gap
final int operations = 1024 - scaledRandomIntBetween(0, 1024);
final Result result = indexOnReplicaWithGaps(indexShard, operations, Math.toIntExact(SequenceNumbersService.NO_OPS_PERFORMED));
final Result result = indexOnReplicaWithGaps(indexShard, operations, Math.toIntExact(SequenceNumbers.NO_OPS_PERFORMED));
final int maxSeqNo = result.maxSeqNo;
final boolean gap = result.gap;
@ -592,7 +592,7 @@ public class IndexShardTests extends IndexShardTestCase {
}
};
indexShard.acquireReplicaOperationPermit(primaryTerm - 1, SequenceNumbersService.UNASSIGNED_SEQ_NO, onLockAcquired,
indexShard.acquireReplicaOperationPermit(primaryTerm - 1, SequenceNumbers.UNASSIGNED_SEQ_NO, onLockAcquired,
ThreadPool.Names.INDEX);
assertFalse(onResponse.get());
@ -608,12 +608,12 @@ public class IndexShardTests extends IndexShardTestCase {
final CyclicBarrier barrier = new CyclicBarrier(2);
final long newPrimaryTerm = primaryTerm + 1 + randomInt(20);
if (engineClosed == false) {
assertThat(indexShard.getLocalCheckpoint(), equalTo(SequenceNumbersService.NO_OPS_PERFORMED));
assertThat(indexShard.getGlobalCheckpoint(), equalTo(SequenceNumbersService.NO_OPS_PERFORMED));
assertThat(indexShard.getLocalCheckpoint(), equalTo(SequenceNumbers.NO_OPS_PERFORMED));
assertThat(indexShard.getGlobalCheckpoint(), equalTo(SequenceNumbers.NO_OPS_PERFORMED));
}
final long newGlobalCheckPoint;
if (engineClosed || randomBoolean()) {
newGlobalCheckPoint = SequenceNumbersService.NO_OPS_PERFORMED;
newGlobalCheckPoint = SequenceNumbers.NO_OPS_PERFORMED;
} else {
long localCheckPoint = indexShard.getGlobalCheckpoint() + randomInt(100);
// advance local checkpoint
@ -623,8 +623,8 @@ public class IndexShardTests extends IndexShardTestCase {
newGlobalCheckPoint = randomIntBetween((int) indexShard.getGlobalCheckpoint(), (int) localCheckPoint);
}
final long expectedLocalCheckpoint;
if (newGlobalCheckPoint == SequenceNumbersService.UNASSIGNED_SEQ_NO) {
expectedLocalCheckpoint = SequenceNumbersService.NO_OPS_PERFORMED;
if (newGlobalCheckPoint == SequenceNumbers.UNASSIGNED_SEQ_NO) {
expectedLocalCheckpoint = SequenceNumbers.NO_OPS_PERFORMED;
} else {
expectedLocalCheckpoint = newGlobalCheckPoint;
}
@ -714,18 +714,18 @@ public class IndexShardTests extends IndexShardTestCase {
public void testRestoreLocalCheckpointTrackerFromTranslogOnPromotion() throws IOException, InterruptedException {
final IndexShard indexShard = newStartedShard(false);
final int operations = 1024 - scaledRandomIntBetween(0, 1024);
indexOnReplicaWithGaps(indexShard, operations, Math.toIntExact(SequenceNumbersService.NO_OPS_PERFORMED));
indexOnReplicaWithGaps(indexShard, operations, Math.toIntExact(SequenceNumbers.NO_OPS_PERFORMED));
final long maxSeqNo = indexShard.seqNoStats().getMaxSeqNo();
final long globalCheckpointOnReplica = SequenceNumbersService.UNASSIGNED_SEQ_NO;
final long globalCheckpointOnReplica = SequenceNumbers.UNASSIGNED_SEQ_NO;
randomIntBetween(
Math.toIntExact(SequenceNumbersService.UNASSIGNED_SEQ_NO),
Math.toIntExact(SequenceNumbers.UNASSIGNED_SEQ_NO),
Math.toIntExact(indexShard.getLocalCheckpoint()));
indexShard.updateGlobalCheckpointOnReplica(globalCheckpointOnReplica, "test");
final int globalCheckpoint =
randomIntBetween(
Math.toIntExact(SequenceNumbersService.UNASSIGNED_SEQ_NO),
Math.toIntExact(SequenceNumbers.UNASSIGNED_SEQ_NO),
Math.toIntExact(indexShard.getLocalCheckpoint()));
final CountDownLatch latch = new CountDownLatch(1);
@ -770,17 +770,17 @@ public class IndexShardTests extends IndexShardTestCase {
// most of the time this is large enough that most of the time there will be at least one gap
final int operations = 1024 - scaledRandomIntBetween(0, 1024);
indexOnReplicaWithGaps(indexShard, operations, Math.toIntExact(SequenceNumbersService.NO_OPS_PERFORMED));
indexOnReplicaWithGaps(indexShard, operations, Math.toIntExact(SequenceNumbers.NO_OPS_PERFORMED));
final long globalCheckpointOnReplica =
randomIntBetween(
Math.toIntExact(SequenceNumbersService.UNASSIGNED_SEQ_NO),
Math.toIntExact(SequenceNumbers.UNASSIGNED_SEQ_NO),
Math.toIntExact(indexShard.getLocalCheckpoint()));
indexShard.updateGlobalCheckpointOnReplica(globalCheckpointOnReplica, "test");
final int globalCheckpoint =
randomIntBetween(
Math.toIntExact(SequenceNumbersService.UNASSIGNED_SEQ_NO),
Math.toIntExact(SequenceNumbers.UNASSIGNED_SEQ_NO),
Math.toIntExact(indexShard.getLocalCheckpoint()));
final CountDownLatch latch = new CountDownLatch(1);
indexShard.acquireReplicaOperationPermit(
@ -801,9 +801,9 @@ public class IndexShardTests extends IndexShardTestCase {
ThreadPool.Names.SAME);
latch.await();
if (globalCheckpointOnReplica == SequenceNumbersService.UNASSIGNED_SEQ_NO
&& globalCheckpoint == SequenceNumbersService.UNASSIGNED_SEQ_NO) {
assertThat(indexShard.getLocalCheckpoint(), equalTo(SequenceNumbersService.NO_OPS_PERFORMED));
if (globalCheckpointOnReplica == SequenceNumbers.UNASSIGNED_SEQ_NO
&& globalCheckpoint == SequenceNumbers.UNASSIGNED_SEQ_NO) {
assertThat(indexShard.getLocalCheckpoint(), equalTo(SequenceNumbers.NO_OPS_PERFORMED));
} else {
assertThat(indexShard.getLocalCheckpoint(), equalTo(Math.max(globalCheckpoint, globalCheckpointOnReplica)));
}

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

@ -24,7 +24,7 @@ import org.elasticsearch.index.engine.Engine;
import org.elasticsearch.index.engine.InternalEngineTests;
import org.elasticsearch.index.mapper.ParsedDocument;
import org.elasticsearch.index.mapper.Uid;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.test.ESTestCase;
import java.util.ArrayList;
@ -138,7 +138,7 @@ public class IndexingOperationListenerTests extends ESTestCase{
ParsedDocument doc = InternalEngineTests.createParsedDoc("1", null);
Engine.Delete delete = new Engine.Delete("test", "1", new Term("_uid", Uid.createUidAsBytes(doc.type(), doc.id())));
Engine.Index index = new Engine.Index(new Term("_uid", Uid.createUidAsBytes(doc.type(), doc.id())), doc);
compositeListener.postDelete(randomShardId, delete, new Engine.DeleteResult(1, SequenceNumbersService.UNASSIGNED_SEQ_NO, true));
compositeListener.postDelete(randomShardId, delete, new Engine.DeleteResult(1, SequenceNumbers.UNASSIGNED_SEQ_NO, true));
assertEquals(0, preIndex.get());
assertEquals(0, postIndex.get());
assertEquals(0, postIndexException.get());
@ -162,7 +162,7 @@ public class IndexingOperationListenerTests extends ESTestCase{
assertEquals(2, postDelete.get());
assertEquals(2, postDeleteException.get());
compositeListener.postIndex(randomShardId, index, new Engine.IndexResult(0, SequenceNumbersService.UNASSIGNED_SEQ_NO, false));
compositeListener.postIndex(randomShardId, index, new Engine.IndexResult(0, SequenceNumbers.UNASSIGNED_SEQ_NO, false));
assertEquals(0, preIndex.get());
assertEquals(2, postIndex.get());
assertEquals(0, postIndexException.get());

63
core/src/test/java/org/elasticsearch/index/translog/TranslogTests.java
View File

@ -34,7 +34,6 @@ import org.apache.lucene.store.MockDirectoryWrapper;
import org.apache.lucene.util.IOUtils;
import org.apache.lucene.util.LineFileDocs;
import org.apache.lucene.util.LuceneTestCase;
import org.elasticsearch.Version;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.common.Randomness;
import org.elasticsearch.common.bytes.BytesArray;
@ -64,7 +63,7 @@ import org.elasticsearch.index.mapper.Uid;
import org.elasticsearch.index.mapper.UidFieldMapper;
import org.elasticsearch.index.seqno.LocalCheckpointTracker;
import org.elasticsearch.index.seqno.LocalCheckpointTrackerTests;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.index.translog.Translog.Location;
import org.elasticsearch.test.ESTestCase;
@ -146,7 +145,7 @@ public class TranslogTests extends ESTestCase {
protected Translog createTranslog(TranslogConfig config, String translogUUID) throws IOException {
return new Translog(config, translogUUID, createTranslogDeletionPolicy(config.getIndexSettings()),
() -> SequenceNumbersService.UNASSIGNED_SEQ_NO);
() -> SequenceNumbers.UNASSIGNED_SEQ_NO);
}
private void markCurrentGenAsCommitted(Translog translog) throws IOException {
@ -188,7 +187,7 @@ public class TranslogTests extends ESTestCase {
}
private Translog create(Path path) throws IOException {
globalCheckpoint = new AtomicLong(SequenceNumbersService.UNASSIGNED_SEQ_NO);
globalCheckpoint = new AtomicLong(SequenceNumbers.UNASSIGNED_SEQ_NO);
final TranslogConfig translogConfig = getTranslogConfig(path);
final TranslogDeletionPolicy deletionPolicy = createTranslogDeletionPolicy(translogConfig.getIndexSettings());
return new Translog(translogConfig, null, deletionPolicy, () -> globalCheckpoint.get());
@ -771,7 +770,7 @@ public class TranslogTests extends ESTestCase {
final AtomicBoolean run = new AtomicBoolean(true);
final Object flushMutex = new Object();
final AtomicLong lastCommittedLocalCheckpoint = new AtomicLong(SequenceNumbersService.NO_OPS_PERFORMED);
final AtomicLong lastCommittedLocalCheckpoint = new AtomicLong(SequenceNumbers.NO_OPS_PERFORMED);
final LocalCheckpointTracker tracker = LocalCheckpointTrackerTests.createEmptyTracker();
final TranslogDeletionPolicy deletionPolicy = translog.getDeletionPolicy();
// any errors on threads
@ -1102,10 +1101,10 @@ public class TranslogTests extends ESTestCase {
out.writeInt(i);
long seqNo;
do {
seqNo = opsHaveValidSequenceNumbers ? randomNonNegativeLong() : SequenceNumbersService.UNASSIGNED_SEQ_NO;
seqNo = opsHaveValidSequenceNumbers ? randomNonNegativeLong() : SequenceNumbers.UNASSIGNED_SEQ_NO;
opsHaveValidSequenceNumbers = opsHaveValidSequenceNumbers || !rarely();
} while (seenSeqNos.contains(seqNo));
if (seqNo != SequenceNumbersService.UNASSIGNED_SEQ_NO) {
if (seqNo != SequenceNumbers.UNASSIGNED_SEQ_NO) {
seenSeqNos.add(seqNo);
}
writer.add(new BytesArray(bytes), seqNo);
@ -1120,8 +1119,8 @@ public class TranslogTests extends ESTestCase {
final int value = buffer.getInt();
assertEquals(i, value);
}
final long minSeqNo = seenSeqNos.stream().min(Long::compareTo).orElse(SequenceNumbersService.NO_OPS_PERFORMED);
final long maxSeqNo = seenSeqNos.stream().max(Long::compareTo).orElse(SequenceNumbersService.NO_OPS_PERFORMED);
final long minSeqNo = seenSeqNos.stream().min(Long::compareTo).orElse(SequenceNumbers.NO_OPS_PERFORMED);
final long maxSeqNo = seenSeqNos.stream().max(Long::compareTo).orElse(SequenceNumbers.NO_OPS_PERFORMED);
assertThat(reader.getCheckpoint().minSeqNo, equalTo(minSeqNo));
assertThat(reader.getCheckpoint().maxSeqNo, equalTo(maxSeqNo));
@ -1211,7 +1210,7 @@ public class TranslogTests extends ESTestCase {
assertNull(snapshot.next());
}
} else {
translog = new Translog(config, translogGeneration.translogUUID, translog.getDeletionPolicy(), () -> SequenceNumbersService.UNASSIGNED_SEQ_NO);
translog = new Translog(config, translogGeneration.translogUUID, translog.getDeletionPolicy(), () -> SequenceNumbers.UNASSIGNED_SEQ_NO);
assertEquals("lastCommitted must be 1 less than current", translogGeneration.translogFileGeneration + 1, translog.currentFileGeneration());
assertFalse(translog.syncNeeded());
try (Translog.Snapshot snapshot = translog.newSnapshotFromGen(translogGeneration.translogFileGeneration)) {
@ -1249,7 +1248,7 @@ public class TranslogTests extends ESTestCase {
TranslogConfig config = translog.getConfig();
final String translogUUID = translog.getTranslogUUID();
final TranslogDeletionPolicy deletionPolicy = translog.getDeletionPolicy();
try (Translog translog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbersService.UNASSIGNED_SEQ_NO)) {
try (Translog translog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbers.UNASSIGNED_SEQ_NO)) {
assertNotNull(translogGeneration);
assertEquals("lastCommitted must be 2 less than current - we never finished the commit", translogGeneration.translogFileGeneration + 2, translog.currentFileGeneration());
assertFalse(translog.syncNeeded());
@ -1263,7 +1262,7 @@ public class TranslogTests extends ESTestCase {
}
}
if (randomBoolean()) { // recover twice
try (Translog translog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbersService.UNASSIGNED_SEQ_NO)) {
try (Translog translog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbers.UNASSIGNED_SEQ_NO)) {
assertNotNull(translogGeneration);
assertEquals("lastCommitted must be 3 less than current - we never finished the commit and run recovery twice", translogGeneration.translogFileGeneration + 3, translog.currentFileGeneration());
assertFalse(translog.syncNeeded());
@ -1307,7 +1306,7 @@ public class TranslogTests extends ESTestCase {
final String translogUUID = translog.getTranslogUUID();
final TranslogDeletionPolicy deletionPolicy = translog.getDeletionPolicy();
try (Translog translog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbersService.UNASSIGNED_SEQ_NO)) {
try (Translog translog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbers.UNASSIGNED_SEQ_NO)) {
assertNotNull(translogGeneration);
assertEquals("lastCommitted must be 2 less than current - we never finished the commit", translogGeneration.translogFileGeneration + 2, translog.currentFileGeneration());
assertFalse(translog.syncNeeded());
@ -1322,7 +1321,7 @@ public class TranslogTests extends ESTestCase {
}
if (randomBoolean()) { // recover twice
try (Translog translog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbersService.UNASSIGNED_SEQ_NO)) {
try (Translog translog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbers.UNASSIGNED_SEQ_NO)) {
assertNotNull(translogGeneration);
assertEquals("lastCommitted must be 3 less than current - we never finished the commit and run recovery twice", translogGeneration.translogFileGeneration + 3, translog.currentFileGeneration());
assertFalse(translog.syncNeeded());
@ -1359,11 +1358,11 @@ public class TranslogTests extends ESTestCase {
TranslogConfig config = translog.getConfig();
Path ckp = config.getTranslogPath().resolve(Translog.CHECKPOINT_FILE_NAME);
Checkpoint read = Checkpoint.read(ckp);
Checkpoint corrupted = Checkpoint.emptyTranslogCheckpoint(0, 0, SequenceNumbersService.UNASSIGNED_SEQ_NO, 0);
Checkpoint corrupted = Checkpoint.emptyTranslogCheckpoint(0, 0, SequenceNumbers.UNASSIGNED_SEQ_NO, 0);
Checkpoint.write(FileChannel::open, config.getTranslogPath().resolve(Translog.getCommitCheckpointFileName(read.generation)), corrupted, StandardOpenOption.WRITE, StandardOpenOption.CREATE_NEW);
final String translogUUID = translog.getTranslogUUID();
final TranslogDeletionPolicy deletionPolicy = translog.getDeletionPolicy();
try (Translog ignored = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbersService.UNASSIGNED_SEQ_NO)) {
try (Translog ignored = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbers.UNASSIGNED_SEQ_NO)) {
fail("corrupted");
} catch (IllegalStateException ex) {
assertEquals("Checkpoint file translog-2.ckp already exists but has corrupted content expected: Checkpoint{offset=3123, " +
@ -1371,7 +1370,7 @@ public class TranslogTests extends ESTestCase {
"generation=0, minSeqNo=-1, maxSeqNo=-1, globalCheckpoint=-2, minTranslogGeneration=0}", ex.getMessage());
}
Checkpoint.write(FileChannel::open, config.getTranslogPath().resolve(Translog.getCommitCheckpointFileName(read.generation)), read, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING);
try (Translog translog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbersService.UNASSIGNED_SEQ_NO)) {
try (Translog translog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbers.UNASSIGNED_SEQ_NO)) {
assertNotNull(translogGeneration);
assertEquals("lastCommitted must be 2 less than current - we never finished the commit", translogGeneration.translogFileGeneration + 2, translog.currentFileGeneration());
assertFalse(translog.syncNeeded());
@ -1448,12 +1447,12 @@ public class TranslogTests extends ESTestCase {
final String foreignTranslog = randomRealisticUnicodeOfCodepointLengthBetween(1,
translogGeneration.translogUUID.length());
try {
new Translog(config, foreignTranslog, createTranslogDeletionPolicy(), () -> SequenceNumbersService.UNASSIGNED_SEQ_NO);
new Translog(config, foreignTranslog, createTranslogDeletionPolicy(), () -> SequenceNumbers.UNASSIGNED_SEQ_NO);
fail("translog doesn't belong to this UUID");
} catch (TranslogCorruptedException ex) {
}
this.translog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbersService.UNASSIGNED_SEQ_NO);
this.translog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbers.UNASSIGNED_SEQ_NO);
try (Translog.Snapshot snapshot = this.translog.newSnapshotFromGen(translogGeneration.translogFileGeneration)) {
for (int i = firstUncommitted; i < translogOperations; i++) {
Translog.Operation next = snapshot.next();
@ -1639,7 +1638,7 @@ public class TranslogTests extends ESTestCase {
translog.close(); // we are closed
final String translogUUID = translog.getTranslogUUID();
final TranslogDeletionPolicy deletionPolicy = translog.getDeletionPolicy();
try (Translog tlog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbersService.UNASSIGNED_SEQ_NO)) {
try (Translog tlog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbers.UNASSIGNED_SEQ_NO)) {
assertEquals("lastCommitted must be 1 less than current", translogGeneration.translogFileGeneration + 1, tlog.currentFileGeneration());
assertFalse(tlog.syncNeeded());
@ -1775,7 +1774,7 @@ public class TranslogTests extends ESTestCase {
}
}
try (Translog tlog =
new Translog(config, translogUUID, createTranslogDeletionPolicy(), () -> SequenceNumbersService.UNASSIGNED_SEQ_NO);
new Translog(config, translogUUID, createTranslogDeletionPolicy(), () -> SequenceNumbers.UNASSIGNED_SEQ_NO);
Translog.Snapshot snapshot = tlog.newSnapshot()) {
if (writtenOperations.size() != snapshot.totalOperations()) {
for (int i = 0; i < threadCount; i++) {
@ -1820,7 +1819,7 @@ public class TranslogTests extends ESTestCase {
TranslogConfig config = translog.getConfig();
final TranslogDeletionPolicy deletionPolicy = new TranslogDeletionPolicy(-1, -1);
deletionPolicy.setMinTranslogGenerationForRecovery(comittedGeneration);
translog = new Translog(config, translog.getTranslogUUID(), deletionPolicy, () -> SequenceNumbersService.UNASSIGNED_SEQ_NO);
translog = new Translog(config, translog.getTranslogUUID(), deletionPolicy, () -> SequenceNumbers.UNASSIGNED_SEQ_NO);
assertThat(translog.getMinFileGeneration(), equalTo(1L));
// no trimming done yet, just recovered
for (long gen = 1; gen < translog.currentFileGeneration(); gen++) {
@ -1874,7 +1873,7 @@ public class TranslogTests extends ESTestCase {
}
final TranslogDeletionPolicy deletionPolicy = new TranslogDeletionPolicy(-1, -1);
deletionPolicy.setMinTranslogGenerationForRecovery(comittedGeneration);
try (Translog translog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbersService.UNASSIGNED_SEQ_NO)) {
try (Translog translog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbers.UNASSIGNED_SEQ_NO)) {
// we don't know when things broke exactly
assertThat(translog.getMinFileGeneration(), greaterThanOrEqualTo(1L));
assertThat(translog.getMinFileGeneration(), lessThanOrEqualTo(comittedGeneration));
@ -1922,7 +1921,7 @@ public class TranslogTests extends ESTestCase {
private Translog getFailableTranslog(final FailSwitch fail, final TranslogConfig config, final boolean partialWrites,
final boolean throwUnknownException, String translogUUID,
final TranslogDeletionPolicy deletionPolicy) throws IOException {
return new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbersService.UNASSIGNED_SEQ_NO) {
return new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbers.UNASSIGNED_SEQ_NO) {
@Override
ChannelFactory getChannelFactory() {
final ChannelFactory factory = super.getChannelFactory();
@ -2048,7 +2047,7 @@ public class TranslogTests extends ESTestCase {
translog.add(new Translog.Index("test", "boom", 0, "boom".getBytes(Charset.forName("UTF-8"))));
translog.close();
try {
new Translog(config, translog.getTranslogUUID(), createTranslogDeletionPolicy(), () -> SequenceNumbersService.UNASSIGNED_SEQ_NO) {
new Translog(config, translog.getTranslogUUID(), createTranslogDeletionPolicy(), () -> SequenceNumbers.UNASSIGNED_SEQ_NO) {
@Override
protected TranslogWriter createWriter(long fileGeneration) throws IOException {
throw new MockDirectoryWrapper.FakeIOException();
@ -2101,7 +2100,7 @@ public class TranslogTests extends ESTestCase {
Files.createFile(config.getTranslogPath().resolve("translog-" + (read.generation + 1) + ".tlog"));
try {
Translog tlog = new Translog(config, translog.getTranslogUUID(), translog.getDeletionPolicy(), () -> SequenceNumbersService.UNASSIGNED_SEQ_NO);
Translog tlog = new Translog(config, translog.getTranslogUUID(), translog.getDeletionPolicy(), () -> SequenceNumbers.UNASSIGNED_SEQ_NO);
fail("file already exists?");
} catch (TranslogException ex) {
// all is well
@ -2123,7 +2122,7 @@ public class TranslogTests extends ESTestCase {
Files.createFile(config.getTranslogPath().resolve("translog-" + (read.generation + 1) + ".tlog"));
// we add N+1 and N+2 to ensure we only delete the N+1 file and never jump ahead and wipe without the right condition
Files.createFile(config.getTranslogPath().resolve("translog-" + (read.generation + 2) + ".tlog"));
try (Translog tlog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbersService.UNASSIGNED_SEQ_NO)) {
try (Translog tlog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbers.UNASSIGNED_SEQ_NO)) {
assertFalse(tlog.syncNeeded());
try (Translog.Snapshot snapshot = tlog.newSnapshot()) {
for (int i = 0; i < 1; i++) {
@ -2136,7 +2135,7 @@ public class TranslogTests extends ESTestCase {
}
try {
Translog tlog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbersService.UNASSIGNED_SEQ_NO);
Translog tlog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbers.UNASSIGNED_SEQ_NO);
fail("file already exists?");
} catch (TranslogException ex) {
// all is well
@ -2239,8 +2238,8 @@ public class TranslogTests extends ESTestCase {
fail.failNever(); // we don't wanna fail here but we might since we write a new checkpoint and create a new tlog file
TranslogDeletionPolicy deletionPolicy = createTranslogDeletionPolicy();
deletionPolicy.setMinTranslogGenerationForRecovery(minGenForRecovery);
try (Translog translog = new Translog(config, generationUUID, deletionPolicy, () -> SequenceNumbersService.UNASSIGNED_SEQ_NO);
Translog.Snapshot snapshot = translog.newSnapshotFromGen(minGenForRecovery)) {
try (Translog translog = new Translog(config, generationUUID, deletionPolicy, () -> SequenceNumbers.UNASSIGNED_SEQ_NO);
Translog.Snapshot snapshot = translog.newSnapshotFromGen(minGenForRecovery)) {
assertEquals(syncedDocs.size(), snapshot.totalOperations());
for (int i = 0; i < syncedDocs.size(); i++) {
Translog.Operation next = snapshot.next();
@ -2304,14 +2303,14 @@ public class TranslogTests extends ESTestCase {
final String translogUUID = translog.getTranslogUUID();
final TranslogDeletionPolicy deletionPolicy = createTranslogDeletionPolicy(config.getIndexSettings());
translog.close();
translog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbersService.UNASSIGNED_SEQ_NO);
translog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbers.UNASSIGNED_SEQ_NO);
translog.add(new Translog.Index("test", "2", 1, new byte[]{2}));
translog.rollGeneration();
Closeable lock = translog.acquireRetentionLock();
translog.add(new Translog.Index("test", "3", 2, new byte[]{3}));
translog.close();
IOUtils.close(lock);
translog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbersService.UNASSIGNED_SEQ_NO);
translog = new Translog(config, translogUUID, deletionPolicy, () -> SequenceNumbers.UNASSIGNED_SEQ_NO);
}
public static Translog.Location randomTranslogLocation() {

8
core/src/test/java/org/elasticsearch/index/translog/TranslogVersionTests.java
View File

@ -19,7 +19,7 @@
package org.elasticsearch.index.translog;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.test.ESTestCase;
import java.io.IOException;
@ -86,10 +86,10 @@ public class TranslogVersionTests extends ESTestCase {
public TranslogReader openReader(final Path path, final long id) throws IOException {
try (FileChannel channel = FileChannel.open(path, StandardOpenOption.READ)) {
final long minSeqNo = SequenceNumbersService.NO_OPS_PERFORMED;
final long maxSeqNo = SequenceNumbersService.NO_OPS_PERFORMED;
final long minSeqNo = SequenceNumbers.NO_OPS_PERFORMED;
final long maxSeqNo = SequenceNumbers.NO_OPS_PERFORMED;
final Checkpoint checkpoint =
new Checkpoint(Files.size(path), 1, id, minSeqNo, maxSeqNo, SequenceNumbersService.UNASSIGNED_SEQ_NO, id);
new Checkpoint(Files.size(path), 1, id, minSeqNo, maxSeqNo, SequenceNumbers.UNASSIGNED_SEQ_NO, id);
return TranslogReader.open(channel, path, checkpoint, null);
}
}

6
core/src/test/java/org/elasticsearch/indices/recovery/PeerRecoveryTargetServiceTests.java
View File

@ -25,7 +25,7 @@ import org.elasticsearch.common.bytes.BytesArray;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.VersionType;
import org.elasticsearch.index.mapper.SourceToParse;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.shard.IndexShard;
import org.elasticsearch.index.shard.IndexShardTestCase;
import org.elasticsearch.index.shard.ShardId;
@ -73,7 +73,7 @@ public class PeerRecoveryTargetServiceTests extends IndexShardTestCase {
translogLocation.set(replica.getTranslog().location());
assertThat(PeerRecoveryTargetService.getStartingSeqNo(recoveryTarget), equalTo(SequenceNumbersService.UNASSIGNED_SEQ_NO));
assertThat(PeerRecoveryTargetService.getStartingSeqNo(recoveryTarget), equalTo(SequenceNumbers.UNASSIGNED_SEQ_NO));
final Translog translog = replica.getTranslog();
translogLocation.set(
@ -87,7 +87,7 @@ public class PeerRecoveryTargetServiceTests extends IndexShardTestCase {
translogLocation.set(replica.getTranslog().location());
// commit is not good, global checkpoint is below max
assertThat(PeerRecoveryTargetService.getStartingSeqNo(recoveryTarget), equalTo(SequenceNumbersService.UNASSIGNED_SEQ_NO));
assertThat(PeerRecoveryTargetService.getStartingSeqNo(recoveryTarget), equalTo(SequenceNumbers.UNASSIGNED_SEQ_NO));
translogLocation.set(
writeTranslog(replica.shardId(), translog.getTranslogUUID(), translog.currentFileGeneration(), maxSeqNo));

20
core/src/test/java/org/elasticsearch/indices/recovery/RecoverySourceHandlerTests.java
View File

@ -56,7 +56,7 @@ import org.elasticsearch.index.mapper.SeqNoFieldMapper;
import org.elasticsearch.index.mapper.Uid;
import org.elasticsearch.index.mapper.UidFieldMapper;
import org.elasticsearch.index.seqno.SeqNoStats;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.shard.IndexShard;
import org.elasticsearch.index.shard.IndexShardRelocatedException;
import org.elasticsearch.index.shard.IndexShardState;
@ -106,7 +106,7 @@ public class RecoverySourceHandlerTests extends ESTestCase {
null,
randomBoolean(),
randomNonNegativeLong(),
randomBoolean() ? SequenceNumbersService.UNASSIGNED_SEQ_NO : randomNonNegativeLong());
randomBoolean() ? SequenceNumbers.UNASSIGNED_SEQ_NO : randomNonNegativeLong());
Store store = newStore(createTempDir());
RecoverySourceHandler handler = new RecoverySourceHandler(null, null, request,
recoverySettings.getChunkSize().bytesAsInt(), Settings.EMPTY);
@ -154,7 +154,7 @@ public class RecoverySourceHandlerTests extends ESTestCase {
public void testSendSnapshotSendsOps() throws IOException {
final RecoverySettings recoverySettings = new RecoverySettings(Settings.EMPTY, service);
final int fileChunkSizeInBytes = recoverySettings.getChunkSize().bytesAsInt();
final long startingSeqNo = randomBoolean() ? SequenceNumbersService.UNASSIGNED_SEQ_NO : randomIntBetween(0, 16);
final long startingSeqNo = randomBoolean() ? SequenceNumbers.UNASSIGNED_SEQ_NO : randomIntBetween(0, 16);
final StartRecoveryRequest request = new StartRecoveryRequest(
shardId,
null,
@ -163,7 +163,7 @@ public class RecoverySourceHandlerTests extends ESTestCase {
null,
randomBoolean(),
randomNonNegativeLong(),
randomBoolean() ? SequenceNumbersService.UNASSIGNED_SEQ_NO : randomNonNegativeLong());
randomBoolean() ? SequenceNumbers.UNASSIGNED_SEQ_NO : randomNonNegativeLong());
final IndexShard shard = mock(IndexShard.class);
when(shard.state()).thenReturn(IndexShardState.STARTED);
final RecoveryTargetHandler recoveryTarget = mock(RecoveryTargetHandler.class);
@ -173,7 +173,7 @@ public class RecoverySourceHandlerTests extends ESTestCase {
final int initialNumberOfDocs = randomIntBetween(16, 64);
for (int i = 0; i < initialNumberOfDocs; i++) {
final Engine.Index index = getIndex(Integer.toString(i));
operations.add(new Translog.Index(index, new Engine.IndexResult(1, SequenceNumbersService.UNASSIGNED_SEQ_NO, true)));
operations.add(new Translog.Index(index, new Engine.IndexResult(1, SequenceNumbers.UNASSIGNED_SEQ_NO, true)));
}
final int numberOfDocsWithValidSequenceNumbers = randomIntBetween(16, 64);
for (int i = initialNumberOfDocs; i < initialNumberOfDocs + numberOfDocsWithValidSequenceNumbers; i++) {
@ -199,7 +199,7 @@ public class RecoverySourceHandlerTests extends ESTestCase {
return operations.get(counter++);
}
});
if (startingSeqNo == SequenceNumbersService.UNASSIGNED_SEQ_NO) {
if (startingSeqNo == SequenceNumbers.UNASSIGNED_SEQ_NO) {
assertThat(result.totalOperations, equalTo(initialNumberOfDocs + numberOfDocsWithValidSequenceNumbers));
} else {
assertThat(result.totalOperations, equalTo(Math.toIntExact(numberOfDocsWithValidSequenceNumbers - startingSeqNo)));
@ -237,7 +237,7 @@ public class RecoverySourceHandlerTests extends ESTestCase {
null,
randomBoolean(),
randomNonNegativeLong(),
randomBoolean() ? SequenceNumbersService.UNASSIGNED_SEQ_NO : 0L);
randomBoolean() ? SequenceNumbers.UNASSIGNED_SEQ_NO : 0L);
Path tempDir = createTempDir();
Store store = newStore(tempDir, false);
AtomicBoolean failedEngine = new AtomicBoolean(false);
@ -307,7 +307,7 @@ public class RecoverySourceHandlerTests extends ESTestCase {
null,
randomBoolean(),
randomNonNegativeLong(),
randomBoolean() ? SequenceNumbersService.UNASSIGNED_SEQ_NO : 0L);
randomBoolean() ? SequenceNumbers.UNASSIGNED_SEQ_NO : 0L);
Path tempDir = createTempDir();
Store store = newStore(tempDir, false);
AtomicBoolean failedEngine = new AtomicBoolean(false);
@ -373,7 +373,7 @@ public class RecoverySourceHandlerTests extends ESTestCase {
null,
false,
randomNonNegativeLong(),
attemptSequenceNumberBasedRecovery ? randomNonNegativeLong() : SequenceNumbersService.UNASSIGNED_SEQ_NO);
attemptSequenceNumberBasedRecovery ? randomNonNegativeLong() : SequenceNumbers.UNASSIGNED_SEQ_NO);
final IndexShard shard = mock(IndexShard.class);
when(shard.seqNoStats()).thenReturn(mock(SeqNoStats.class));
when(shard.segmentStats(anyBoolean())).thenReturn(mock(SegmentsStats.class));
@ -412,7 +412,7 @@ public class RecoverySourceHandlerTests extends ESTestCase {
@Override
long phase2(long startingSeqNo, Translog.Snapshot snapshot) throws IOException {
phase2Called.set(true);
return SequenceNumbersService.UNASSIGNED_SEQ_NO;
return SequenceNumbers.UNASSIGNED_SEQ_NO;
}
};

6
core/src/test/java/org/elasticsearch/indices/recovery/StartRecoveryRequestTests.java
View File

@ -24,7 +24,7 @@ import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.UUIDs;
import org.elasticsearch.common.io.stream.InputStreamStreamInput;
import org.elasticsearch.common.io.stream.OutputStreamStreamOutput;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.index.store.Store;
import org.elasticsearch.test.ESTestCase;
@ -49,7 +49,7 @@ public class StartRecoveryRequestTests extends ESTestCase {
Store.MetadataSnapshot.EMPTY,
randomBoolean(),
randomNonNegativeLong(),
randomBoolean() ? SequenceNumbersService.UNASSIGNED_SEQ_NO : randomNonNegativeLong());
randomBoolean() ? SequenceNumbers.UNASSIGNED_SEQ_NO : randomNonNegativeLong());
final ByteArrayOutputStream outBuffer = new ByteArrayOutputStream();
final OutputStreamStreamOutput out = new OutputStreamStreamOutput(outBuffer);
@ -72,7 +72,7 @@ public class StartRecoveryRequestTests extends ESTestCase {
if (targetNodeVersion.onOrAfter(Version.V_6_0_0_alpha1)) {
assertThat(outRequest.startingSeqNo(), equalTo(inRequest.startingSeqNo()));
} else {
assertThat(SequenceNumbersService.UNASSIGNED_SEQ_NO, equalTo(inRequest.startingSeqNo()));
assertThat(SequenceNumbers.UNASSIGNED_SEQ_NO, equalTo(inRequest.startingSeqNo()));
}
}

4
core/src/test/java/org/elasticsearch/recovery/RelocationIT.java
View File

@ -45,7 +45,7 @@ import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.env.NodeEnvironment;
import org.elasticsearch.index.seqno.SeqNoStats;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.shard.IndexEventListener;
import org.elasticsearch.index.shard.IndexShard;
import org.elasticsearch.index.shard.IndexShardState;
@ -119,7 +119,7 @@ public class RelocationIT extends ESIntegTestCase {
ShardStats primary = maybePrimary.get();
final SeqNoStats primarySeqNoStats = primary.getSeqNoStats();
assertThat(primary.getShardRouting() + " should have set the global checkpoint",
primarySeqNoStats.getGlobalCheckpoint(), not(equalTo(SequenceNumbersService.UNASSIGNED_SEQ_NO)));
primarySeqNoStats.getGlobalCheckpoint(), not(equalTo(SequenceNumbers.UNASSIGNED_SEQ_NO)));
for (ShardStats shardStats : indexShardStats) {
final SeqNoStats seqNoStats = shardStats.getSeqNoStats();
assertThat(shardStats.getShardRouting() + " local checkpoint mismatch",

4
test/framework/src/main/java/org/elasticsearch/index/shard/IndexShardTestCase.java
View File

@ -58,7 +58,7 @@ import org.elasticsearch.index.mapper.MapperService;
import org.elasticsearch.index.mapper.Mapping;
import org.elasticsearch.index.mapper.SourceToParse;
import org.elasticsearch.index.mapper.Uid;
import org.elasticsearch.index.seqno.SequenceNumbersService;
import org.elasticsearch.index.seqno.SequenceNumbers;
import org.elasticsearch.index.similarity.SimilarityService;
import org.elasticsearch.index.store.DirectoryService;
import org.elasticsearch.index.store.Store;
@ -440,7 +440,7 @@ public abstract class IndexShardTestCase extends ESTestCase {
if (snapshot.size() > 0) {
startingSeqNo = PeerRecoveryTargetService.getStartingSeqNo(recoveryTarget);
} else {
startingSeqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO;
startingSeqNo = SequenceNumbers.UNASSIGNED_SEQ_NO;
}
final StartRecoveryRequest request = new StartRecoveryRequest(replica.shardId(), targetAllocationId,