/**
 * EsType_updateメソッドで初回にVersionConflictEngineExceptionが投げられた場合のテスト.
 */
@Test
public void EsType_updateメソッドで初回にVersionConflictEngineExceptionが投げられた場合のテスト() {
    PowerMockito.mockStatic(EsClientException.class);
    EsTypeImpl esTypeObject = Mockito.spy(new EsTypeImpl("dummy", "Test", "TestRoutingId", 0, 0, null));

    // EsType#asyncIndex()が呼ばれた場合に、VersionConflictEngineExceptionを投げる。
    // 送出する例外オブジェクトのモックを作成
    VersionConflictEngineException toBeThrown = Mockito.mock(VersionConflictEngineException.class);
    Mockito.doThrow(toBeThrown)
            .when(esTypeObject)
            .asyncIndex(Mockito.anyString(), Mockito.anyMapOf(String.class, Object.class),
                    (OpType) Mockito.anyObject(), Mockito.anyLong());
    // メソッド呼び出し
    try {
        esTypeObject.update("dummyId", null, 1);
        fail("EsClientException should be thrown.");
    } catch (EsClientException.EsVersionConflictException e) {
        assertTrue(e.getCause() instanceof VersionConflictEngineException);
    }
}
 

/**
 * EsType_deleteメソッドで初回にVersionConflictEngineExceptionが投げられた場合のテスト.
 */
@Test
public void EsType_deleteメソッドで初回にVersionConflictEngineExceptionが投げられた場合のテスト() {
    PowerMockito.mockStatic(EsClientException.class);
    EsTypeImpl esTypeObject = Mockito.spy(new EsTypeImpl("dummy", "Test", "TestRoutingId", 0, 0, null));

    // EsType#asyncDelete()が呼ばれた場合に、VersionConflictEngineExceptionを投げる。
    // 送出する例外オブジェクトのモックを作成
    VersionConflictEngineException toBeThrown = Mockito.mock(VersionConflictEngineException.class);
    Mockito.doThrow(toBeThrown)
            .when(esTypeObject)
            .asyncDelete(Mockito.anyString(), Mockito.anyLong());
    // メソッド呼び出し
    try {
        esTypeObject.delete("dummyId", 1);
        fail("EsClientException should be thrown.");
    } catch (EsClientException.EsVersionConflictException e) {
        assertTrue(e.getCause() instanceof VersionConflictEngineException);
    }
}
 

/**
 * EsType_updateメソッドで初回にVersionConflictEngineExceptionが投げられた場合のテスト.
 */
@Test
public void EsType_updateメソッドで初回にVersionConflictEngineExceptionが投げられた場合のテスト() {
    PowerMockito.mockStatic(EsClientException.class);
    EsTypeImpl esTypeObject = Mockito.spy(new EsTypeImpl("dummy", "Test", "TestRoutingId", 0, 0, null));

    // EsType#asyncIndex()が呼ばれた場合に、VersionConflictEngineExceptionを投げる。
    // 送出する例外オブジェクトのモックを作成
    VersionConflictEngineException toBeThrown = Mockito.mock(VersionConflictEngineException.class);
    Mockito.doThrow(toBeThrown)
            .when(esTypeObject)
            .asyncIndex(Mockito.anyString(), Mockito.anyMapOf(String.class, Object.class),
                    (OpType) Mockito.anyObject(), Mockito.anyLong());
    // メソッド呼び出し
    try {
        esTypeObject.update("dummyId", null, 1);
        fail("EsClientException should be thrown.");
    } catch (EsClientException.EsVersionConflictException e) {
        assertTrue(e.getCause() instanceof VersionConflictEngineException);
    }
}
 

/**
 * EsType_deleteメソッドで初回にVersionConflictEngineExceptionが投げられた場合のテスト.
 */
@Test
public void EsType_deleteメソッドで初回にVersionConflictEngineExceptionが投げられた場合のテスト() {
    PowerMockito.mockStatic(EsClientException.class);
    EsTypeImpl esTypeObject = Mockito.spy(new EsTypeImpl("dummy", "Test", "TestRoutingId", 0, 0, null));

    // EsType#asyncDelete()が呼ばれた場合に、VersionConflictEngineExceptionを投げる。
    // 送出する例外オブジェクトのモックを作成
    VersionConflictEngineException toBeThrown = Mockito.mock(VersionConflictEngineException.class);
    Mockito.doThrow(toBeThrown)
            .when(esTypeObject)
            .asyncDelete(Mockito.anyString(), Mockito.anyLong());
    // メソッド呼び出し
    try {
        esTypeObject.delete("dummyId", 1);
        fail("EsClientException should be thrown.");
    } catch (EsClientException.EsVersionConflictException e) {
        assertTrue(e.getCause() instanceof VersionConflictEngineException);
    }
}
 

private static Doc getDocument(IndexShard indexShard, String id, long version, long seqNo, long primaryTerm) {
    // when sequence versioning is used, this lookup will throw VersionConflictEngineException
    Doc doc = PKLookupOperation.lookupDoc(indexShard, id, Versions.MATCH_ANY, VersionType.INTERNAL, seqNo, primaryTerm);
    if (doc == null) {
        throw new DocumentMissingException(indexShard.shardId(), Constants.DEFAULT_MAPPING_TYPE, id);
    }
    if (doc.getSource() == null) {
        throw new DocumentSourceMissingException(indexShard.shardId(), Constants.DEFAULT_MAPPING_TYPE, id);
    }
    if (version != Versions.MATCH_ANY && version != doc.getVersion()) {
        throw new VersionConflictEngineException(
            indexShard.shardId(),
            id,
            "Requested version: " + version + " but got version: " + doc.getVersion());
    }
    return doc;
}
 

@Test
public void testExceptionWhileProcessingItemsNotContinueOnError() throws Exception {
    ShardId shardId = new ShardId(TABLE_IDENT.indexNameOrAlias(), charactersIndexUUID, 0);
    ShardUpsertRequest request = new ShardUpsertRequest.Builder(
        DUMMY_SESSION_INFO,
        TimeValue.timeValueSeconds(30),
        DuplicateKeyAction.UPDATE_OR_FAIL,
        false,
        null,
        new Reference[]{ID_REF},
        null,
        UUID.randomUUID(),
        false
    ).newRequest(shardId);
    request.add(1, new ShardUpsertRequest.Item("1", null, new Object[]{1}, null, null, null));

    TransportWriteAction.WritePrimaryResult<ShardUpsertRequest, ShardResponse> result =
        transportShardUpsertAction.processRequestItems(indexShard, request, new AtomicBoolean(false));

    assertThat(result.finalResponseIfSuccessful.failure(), instanceOf(VersionConflictEngineException.class));
}
 

@Override
public void onFailure(Throwable e) {
    e = Exceptions.unwrap(e); // unwrap to get rid of RemoteTransportException
    if (e instanceof VersionConflictEngineException) {
        // treat version conflict as rows affected = 0
        result.set(TaskResult.ZERO);
    } else {
        result.setException(e);
    }
}
 

protected boolean isConflictException(Throwable e) {
    Throwable cause = ExceptionsHelper.unwrapCause(e);
    // on version conflict or document missing, it means
    // that a new change has crept into the replica, and it's fine
    if (cause instanceof VersionConflictEngineException) {
        return true;
    }
    if (cause instanceof DocumentAlreadyExistsException) {
        return true;
    }
    return false;
}
 

protected <T> Transform<Throwable, T> handleVersionConflict(
    Provider<T> emptyProvider, Runnable reportWriteDroppedByDuplicate
) {
    return throwable -> {
        if (ExceptionUtils.getRootCause(throwable) instanceof VersionConflictEngineException) {
            // Index request rejected, document already exists. That's ok, return success.
            reportWriteDroppedByDuplicate.run();
            return emptyProvider.get();
        }
        throw new RuntimeException(throwable);
    };
}
 

@Override
boolean isParticularError(ElasticsearchException e) {
    return e instanceof IndexNotFoundException
            || e.getCause() instanceof IndexNotFoundException
            || e instanceof VersionConflictEngineException
            || e instanceof MapperParsingException;
}
 

@Override
IndexResponse onParticularError(ElasticsearchException e) {
    if (e instanceof IndexNotFoundException || e.getCause() instanceof IndexNotFoundException) {
        throw new EsClientException.EsIndexMissingException(e);
    }
    if (e instanceof VersionConflictEngineException) {
        throw new EsClientException.EsVersionConflictException(e);
    }
    if (e instanceof MapperParsingException) {
        throw new EsClientException.EsSchemaMismatchException(e);
    }
    throw e;
}
 

@Override
DeleteResponse onParticularError(ElasticsearchException e) {
    if (e instanceof IndexNotFoundException || e.getCause() instanceof IndexNotFoundException) {
        throw new EsClientException.EsIndexMissingException(e);
    }
    if (e instanceof VersionConflictEngineException) {
        throw new EsClientException.EsVersionConflictException(e);
    }
    throw e;
}
 

@Override
boolean isParticularError(ElasticsearchException e) {
    return e instanceof IndexMissingException
            || e.getCause() instanceof IndexMissingException
            || e instanceof VersionConflictEngineException
            || e instanceof MapperParsingException;
}
 

@Override
IndexResponse onParticularError(ElasticsearchException e) {
    if (e instanceof IndexMissingException || e.getCause() instanceof IndexMissingException) {
        throw new EsClientException.EsIndexMissingException(e);
    }
    if (e instanceof VersionConflictEngineException) {
        throw new EsClientException.EsVersionConflictException(e);
    }
    if (e instanceof MapperParsingException) {
        throw new EsClientException.EsSchemaMismatchException(e);
    }
    throw e;
}
 

@Override
DeleteResponse onParticularError(ElasticsearchException e) {
    if (e instanceof IndexMissingException || e.getCause() instanceof IndexMissingException) {
        throw new EsClientException.EsIndexMissingException(e);
    }
    if (e instanceof VersionConflictEngineException) {
        throw new EsClientException.EsVersionConflictException(e);
    }
    throw e;
}
 

boolean ignoreReplicaException(Throwable e) {
    if (TransportActions.isShardNotAvailableException(e)) {
        return true;
    }
    Throwable cause = ExceptionsHelper.unwrapCause(e);
    return cause instanceof VersionConflictEngineException;
}
 

private static void onResponse(ShardResponse.CompressedResult result, ShardResponse response) {
    Exception failure = response.failure();
    if (failure == null) {
        result.update(response);
    } else {
        Throwable t = SQLExceptions.unwrap(failure, e -> e instanceof RuntimeException);
        if (!(t instanceof DocumentMissingException) && !(t instanceof VersionConflictEngineException)) {
            throw new RuntimeException(t);
        }
    }
}
 

@Override
protected IndexItemResponse insert(ShardUpsertRequest request,
                                   ShardUpsertRequest.Item item,
                                   IndexShard indexShard,
                                   boolean isRetry,
                                   @Nullable ReturnValueGen returnGen,
                                   @Nullable InsertSourceGen insertSourceGen) throws Exception {
    throw new VersionConflictEngineException(
        indexShard.shardId(),
        item.id(),
        "document with id: " + item.id() + " already exists in '" + request.shardId().getIndexName() + '\'');
}
 

@Override
protected ShardResponse processRequestItems(ShardId shardId, ShardDeleteRequest request, AtomicBoolean killed) throws InterruptedException {
    ShardResponse shardResponse = new ShardResponse();
    IndexService indexService = indicesService.indexServiceSafe(request.index());
    IndexShard indexShard = indexService.shardSafe(shardId.id());
    for (int i = 0; i < request.itemIndices().size(); i++) {
        int location = request.itemIndices().get(i);
        ShardDeleteRequest.Item item = request.items().get(i);
        if (killed.get()) {
            // set failure on response, mark current item and skip all next items.
            // this way replica operation will be executed, but only items already processed here
            // will be processed on the replica
            request.skipFromLocation(location);
            shardResponse.failure(new InterruptedException(JobKilledException.MESSAGE));
            break;
        }
        try {
            boolean found = shardDeleteOperationOnPrimary(request, item, indexShard);
            if (found) {
                logger.debug("{} successfully deleted [{}]/[{}]", request.shardId(), request.type(), item.id());
                shardResponse.add(location);
            } else {
                logger.debug("{} failed to execute delete for [{}]/[{}], doc not found",
                        request.shardId(), request.type(), item.id());
                shardResponse.add(location,
                        new ShardResponse.Failure(
                                item.id(),
                                "Document not found while deleting",
                                false));

            }
        } catch (Throwable t) {
            if (!TransportActions.isShardNotAvailableException(t)) {
                throw t;
            } else {
                logger.debug("{} failed to execute delete for [{}]/[{}]",
                        t, request.shardId(), request.type(), item.id());
                shardResponse.add(location,
                        new ShardResponse.Failure(
                                item.id(),
                                ExceptionsHelper.detailedMessage(t),
                                (t instanceof VersionConflictEngineException)));
            }
        }
    }

    return shardResponse;
}
 

@Override
protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
    String store = indexName(request);
    Long expectedVersion = null;
    if (request.hasParam("version")) {
        expectedVersion = request.paramAsLong("version", -1);
        if (expectedVersion <= 0) {
            throw new IllegalArgumentException("version must be a strictly positive long value");
        }
    }
    String routing = request.param("routing");
    ParserState state = new ParserState();
    request.withContentOrSourceParamParserOrNull((p) -> ParserState.parse(p, state));
    CreateModelFromSetRequestBuilder builder = new CreateModelFromSetRequestBuilder(client);
    if (expectedVersion != null) {
        builder.withVersion(store, request.param("name"), expectedVersion, state.model.name, state.model.model);
    } else {
        builder.withoutVersion(store, request.param("name"), state.model.name, state.model.model);
    }
    builder.request().setValidation(state.validation);
    builder.routing(routing);
    return (channel) -> builder.execute(ActionListener.wrap(
            response -> new RestStatusToXContentListener<CreateModelFromSetAction.CreateModelFromSetResponse>(channel,
                    (r) -> r.getResponse().getLocation(routing)).onResponse(response),
            (e) -> {
                final Exception exc;
                final RestStatus status;
                if (ExceptionsHelper.unwrap(e, VersionConflictEngineException.class) != null) {
                    exc = new IllegalArgumentException("Element of type [" + StoredLtrModel.TYPE +
                            "] are not updatable, please create a new one instead.");
                    exc.addSuppressed(e);
                    status = RestStatus.METHOD_NOT_ALLOWED;
                } else {
                    exc = e;
                    status = ExceptionsHelper.status(exc);
                }

                try {
                    channel.sendResponse(new BytesRestResponse(channel, status, exc));
                } catch (Exception inner) {
                    inner.addSuppressed(e);
                    logger.error("failed to send failure response", inner);
                }
            }
    ));
}
 

@Override
boolean isParticularError(ElasticsearchException e) {
    return e instanceof IndexNotFoundException
            || e.getCause() instanceof IndexNotFoundException
            || e instanceof VersionConflictEngineException;
}
 

@Override
boolean isParticularError(ElasticsearchException e) {
    return e instanceof IndexMissingException
            || e.getCause() instanceof IndexMissingException
            || e instanceof VersionConflictEngineException;
}
 

public static boolean isDocumentAlreadyExistsException(Throwable e) {
    return e instanceof VersionConflictEngineException
               && e.getMessage().contains("document already exists");
}
 

@Override
protected WritePrimaryResult<ShardUpsertRequest, ShardResponse> processRequestItems(IndexShard indexShard,
                                                                                    ShardUpsertRequest request,
                                                                                    AtomicBoolean killed) {
    ShardResponse shardResponse = new ShardResponse(request.returnValues());
    String indexName = request.index();
    DocTableInfo tableInfo = schemas.getTableInfo(RelationName.fromIndexName(indexName), Operation.INSERT);
    Reference[] insertColumns = request.insertColumns();
    GeneratedColumns.Validation valueValidation = request.validateConstraints()
        ? GeneratedColumns.Validation.VALUE_MATCH
        : GeneratedColumns.Validation.NONE;

    TransactionContext txnCtx = TransactionContext.of(request.sessionSettings());
    InsertSourceGen insertSourceGen = insertColumns == null
        ? null
        : InsertSourceGen.of(txnCtx, functions, tableInfo, indexName, valueValidation, Arrays.asList(insertColumns));

    UpdateSourceGen updateSourceGen = request.updateColumns() == null
        ? null
        : new UpdateSourceGen(functions,
                              txnCtx,
                              tableInfo,
                              request.updateColumns());

    ReturnValueGen returnValueGen = request.returnValues() == null
        ? null
        : new ReturnValueGen(functions, txnCtx, tableInfo, request.returnValues());

    Translog.Location translogLocation = null;
    for (ShardUpsertRequest.Item item : request.items()) {
        int location = item.location();
        if (killed.get()) {
            // set failure on response and skip all next items.
            // this way replica operation will be executed, but only items with a valid source (= was processed on primary)
            // will be processed on the replica
            shardResponse.failure(new InterruptedException());
            break;
        }
        try {
            IndexItemResponse indexItemResponse = indexItem(
                request,
                item,
                indexShard,
                updateSourceGen,
                insertSourceGen,
                returnValueGen
            );
            if (indexItemResponse != null) {
                if (indexItemResponse.translog != null) {
                    shardResponse.add(location);
                    translogLocation = indexItemResponse.translog;
                }
                if (indexItemResponse.returnValues != null) {
                    shardResponse.addResultRows(indexItemResponse.returnValues);
                }
            }
        } catch (Exception e) {
            if (retryPrimaryException(e)) {
                if (e instanceof RuntimeException) {
                    throw (RuntimeException) e;
                }
                throw new RuntimeException(e);
            }
            if (logger.isDebugEnabled()) {
                logger.debug("Failed to execute upsert shardId={} id={} error={}", request.shardId(), item.id(), e);
            }

            // *mark* the item as failed by setting the source to null
            // to prevent the replica operation from processing this concrete item
            item.source(null);

            if (!request.continueOnError()) {
                shardResponse.failure(e);
                break;
            }
            shardResponse.add(location,
                new ShardResponse.Failure(
                    item.id(),
                    userFriendlyCrateExceptionTopOnly(e),
                    (e instanceof VersionConflictEngineException)));
        }
    }
    return new WritePrimaryResult<>(request, shardResponse, translogLocation, null, indexShard);
}
 

@Nullable
private IndexItemResponse indexItem(ShardUpsertRequest request,
                                    ShardUpsertRequest.Item item,
                                    IndexShard indexShard,
                                    @Nullable UpdateSourceGen updateSourceGen,
                                    @Nullable InsertSourceGen insertSourceGen,
                                    @Nullable ReturnValueGen returnValueGen) throws Exception {
    VersionConflictEngineException lastException = null;
    boolean tryInsertFirst = item.insertValues() != null;
    boolean isRetry;
    for (int retryCount = 0; retryCount < MAX_RETRY_LIMIT; retryCount++) {
        try {
            isRetry = retryCount > 0;
            if (tryInsertFirst) {
                return insert(request, item, indexShard, isRetry, returnValueGen, insertSourceGen);
            } else {
                return update(item, indexShard, isRetry, returnValueGen, updateSourceGen);
            }
        } catch (VersionConflictEngineException e) {
            lastException = e;
            if (request.duplicateKeyAction() == DuplicateKeyAction.IGNORE) {
                // on conflict do nothing
                item.source(null);
                return null;
            }
            Symbol[] updateAssignments = item.updateAssignments();
            if (updateAssignments != null && updateAssignments.length > 0) {
                if (tryInsertFirst) {
                    // insert failed, document already exists, try update
                    tryInsertFirst = false;
                    continue;
                } else if (item.retryOnConflict()) {
                    if (logger.isTraceEnabled()) {
                        logger.trace("[{}] VersionConflict, retrying operation for document id={}, version={} retryCount={}",
                            indexShard.shardId(), item.id(), item.version(), retryCount);
                    }
                    continue;
                }
            }
            throw e;
        }
    }
    logger.warn("[{}] VersionConflict for document id={}, version={} exceeded retry limit of {}, will stop retrying",
        indexShard.shardId(), item.id(), item.version(), MAX_RETRY_LIMIT);
    throw lastException;
}