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

cleanup indexing operation listener

This commit is contained in:
Areek Zillur 2016-10-25 09:33:04 -04:00
parent 168946ad5a
commit bb785483ae
7 changed files with 37 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -536,16 +536,16 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
return new Engine.Index(uid, doc, version, versionType, origin, startTime, autoGeneratedIdTimestamp, isRetry);
}
public Engine.IndexResult index(Engine.Index index) {
public Engine.IndexResult index(final Engine.Index index) {
ensureWriteAllowed(index);
Engine engine = getEngine();
return index(engine, index);
}
private Engine.IndexResult index(Engine engine, Engine.Index index) {
private Engine.IndexResult index(final Engine engine, final Engine.Index index) {
active.set(true);
final Engine.IndexResult result;
index = indexingOperationListeners.preIndex(index);
indexingOperationListeners.preIndex(index);
try {
if (logger.isTraceEnabled()) {
logger.trace("index [{}][{}]{}", index.type(), index.id(), index.docs());
@ -585,16 +585,16 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
return new Engine.Delete(type, id, uid, version, versionType, origin, startTime);
}
public Engine.DeleteResult delete(Engine.Delete delete) {
public Engine.DeleteResult delete(final Engine.Delete delete) {
ensureWriteAllowed(delete);
Engine engine = getEngine();
return delete(engine, delete);
}
private Engine.DeleteResult delete(Engine engine, Engine.Delete delete) {
private Engine.DeleteResult delete(final Engine engine, final Engine.Delete delete) {
active.set(true);
final Engine.DeleteResult result;
delete = indexingOperationListeners.preDelete(delete);
indexingOperationListeners.preDelete(delete);
try {
if (logger.isTraceEnabled()) {
logger.trace("delete [{}]", delete.uid().text());

47
core/src/main/java/org/elasticsearch/index/shard/IndexingOperationListener.java
View File

@ -30,44 +30,25 @@ import java.util.List;
*/
public interface IndexingOperationListener {
/**
* Called before the indexing occurs.
*/
default Engine.Index preIndex(Engine.Index operation) {
return operation;
}
/** Called before the indexing occurs */
default void preIndex(Engine.Index operation) {}
/**
* Called after the indexing operation occurred.
*/
/** Called after the indexing operation occurred */
default void postIndex(Engine.Index index, Engine.IndexResult result) {}
/**
* Called after the indexing operation occurred with exception.
*/
/** Called after the indexing operation occurred with exception */
default void postIndex(Engine.Index index, Exception ex) {}
/**
* Called before the delete occurs.
*/
default Engine.Delete preDelete(Engine.Delete delete) {
return delete;
}
/** Called before the delete occurs */
default void preDelete(Engine.Delete delete) {}
/**
* Called after the delete operation occurred.
*/
/** Called after the delete operation occurred */
default void postDelete(Engine.Delete delete, Engine.DeleteResult result) {}
/**
* Called after the delete operation occurred with exception.
*/
/** Called after the delete operation occurred with exception */
default void postDelete(Engine.Delete delete, Exception ex) {}
/**
* A Composite listener that multiplexes calls to each of the listeners methods.
*/
/** A Composite listener that multiplexes calls to each of the listeners methods */
final class CompositeListener implements IndexingOperationListener{
private final List<IndexingOperationListener> listeners;
private final Logger logger;
@ -78,7 +59,7 @@ public interface IndexingOperationListener {
}
@Override
public Engine.Index preIndex(Engine.Index operation) {
public void preIndex(Engine.Index operation) {
assert operation != null;
for (IndexingOperationListener listener : listeners) {
try {
@ -87,12 +68,11 @@ public interface IndexingOperationListener {
logger.warn((Supplier<?>) () -> new ParameterizedMessage("preIndex listener [{}] failed", listener), e);
}
}
return operation;
}
@Override
public void postIndex(Engine.Index index, Engine.IndexResult result) {
assert index != null;
assert index != null && result != null;
for (IndexingOperationListener listener : listeners) {
try {
listener.postIndex(index, result);
@ -116,7 +96,7 @@ public interface IndexingOperationListener {
}
@Override
public Engine.Delete preDelete(Engine.Delete delete) {
public void preDelete(Engine.Delete delete) {
assert delete != null;
for (IndexingOperationListener listener : listeners) {
try {
@ -125,12 +105,11 @@ public interface IndexingOperationListener {
logger.warn((Supplier<?>) () -> new ParameterizedMessage("preDelete listener [{}] failed", listener), e);
}
}
return delete;
}
@Override
public void postDelete(Engine.Delete delete, Engine.DeleteResult result) {
assert delete != null;
assert delete != null && result != null;
for (IndexingOperationListener listener : listeners) {
try {
listener.postDelete(delete, result);

7
core/src/main/java/org/elasticsearch/index/shard/InternalIndexingStats.java
View File

@ -65,12 +65,11 @@ final class InternalIndexingStats implements IndexingOperationListener {
}
@Override
public Engine.Index preIndex(Engine.Index operation) {
public void preIndex(Engine.Index operation) {
if (!operation.origin().isRecovery()) {
totalStats.indexCurrent.inc();
typeStats(operation.type()).indexCurrent.inc();
}
return operation;
}
@Override
@ -96,13 +95,11 @@ final class InternalIndexingStats implements IndexingOperationListener {
}
@Override
public Engine.Delete preDelete(Engine.Delete delete) {
public void preDelete(Engine.Delete delete) {
if (!delete.origin().isRecovery()) {
totalStats.deleteCurrent.inc();
typeStats(delete.type()).deleteCurrent.inc();
}
return delete;
}
@Override

3
core/src/test/java/org/elasticsearch/index/IndexModuleTests.java
View File

@ -233,9 +233,8 @@ public class IndexModuleTests extends ESTestCase {
AtomicBoolean executed = new AtomicBoolean(false);
IndexingOperationListener listener = new IndexingOperationListener() {
@Override
public Engine.Index preIndex(Engine.Index operation) {
public void preIndex(Engine.Index operation) {
executed.set(true);
return operation;
}
};
module.addIndexOperationListener(listener);

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

@ -558,9 +558,8 @@ public class IndexShardTests extends IndexShardTestCase {
shard.close("simon says", true);
shard = reinitShard(shard, new IndexingOperationListener() {
@Override
public Engine.Index preIndex(Engine.Index operation) {
public void preIndex(Engine.Index operation) {
preIndex.incrementAndGet();
return operation;
}
@Override
@ -578,9 +577,8 @@ public class IndexShardTests extends IndexShardTestCase {
}
@Override
public Engine.Delete preDelete(Engine.Delete delete) {
public void preDelete(Engine.Delete delete) {
preDelete.incrementAndGet();
return delete;
}
@Override
@ -1121,9 +1119,8 @@ public class IndexShardTests extends IndexShardTestCase {
final AtomicInteger postDelete = new AtomicInteger();
IndexingOperationListener listener = new IndexingOperationListener() {
@Override
public Engine.Index preIndex(Engine.Index operation) {
public void preIndex(Engine.Index operation) {
preIndex.incrementAndGet();
return operation;
}
@Override
@ -1132,9 +1129,8 @@ public class IndexShardTests extends IndexShardTestCase {
}
@Override
public Engine.Delete preDelete(Engine.Delete delete) {
public void preDelete(Engine.Delete delete) {
preDelete.incrementAndGet();
return delete;
}
@Override

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

@ -40,9 +40,8 @@ public class IndexingOperationListenerTests extends ESTestCase{
AtomicInteger postDeleteException = new AtomicInteger();
IndexingOperationListener listener = new IndexingOperationListener() {
@Override
public Engine.Index preIndex(Engine.Index operation) {
public void preIndex(Engine.Index operation) {
preIndex.incrementAndGet();
return operation;
}
@Override
@ -56,9 +55,8 @@ public class IndexingOperationListenerTests extends ESTestCase{
}
@Override
public Engine.Delete preDelete(Engine.Delete delete) {
public void preDelete(Engine.Delete delete) {
preDelete.incrementAndGet();
return delete;
}
@Override
@ -74,7 +72,7 @@ public class IndexingOperationListenerTests extends ESTestCase{
IndexingOperationListener throwingListener = new IndexingOperationListener() {
@Override
public Engine.Index preIndex(Engine.Index operation) {
public void preIndex(Engine.Index operation) {
throw new RuntimeException();
}
@ -89,7 +87,7 @@ public class IndexingOperationListenerTests extends ESTestCase{
}
@Override
public Engine.Delete preDelete(Engine.Delete delete) {
public void preDelete(Engine.Delete delete) {
throw new RuntimeException();
}

14
modules/reindex/src/test/java/org/elasticsearch/index/reindex/CancelTests.java
View File

@ -199,23 +199,23 @@ public class CancelTests extends ReindexTestCase {
public static class BlockingOperationListener implements IndexingOperationListener {
@Override
public Engine.Index preIndex(Engine.Index index) {
return preCheck(index, index.type());
public void preIndex(Engine.Index index) {
preCheck(index, index.type());
}
@Override
public Engine.Delete preDelete(Engine.Delete delete) {
return preCheck(delete, delete.type());
public void preDelete(Engine.Delete delete) {
preCheck(delete, delete.type());
}
private <T extends Engine.Operation> T preCheck(T operation, String type) {
private void preCheck(Engine.Operation operation, String type) {
if ((TYPE.equals(type) == false) || (operation.origin() != Origin.PRIMARY)) {
return operation;
return;
}
try {
if (ALLOWED_OPERATIONS.tryAcquire(30, TimeUnit.SECONDS)) {
return operation;
return;
}
} catch (InterruptedException e) {
throw new RuntimeException(e);