alexey.rassokhin/tet123
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

DBZ-648 InterrupedException handled by MongoDB

Browse Source
This commit is contained in:
Jiri Pechanec 2018-03-09 12:42:44 +01:00 committed by Gunnar Morling
parent e113853768
commit 2a35eae178
2 changed files with 21 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
debezium-connector-mongodb/src/main/java/io/debezium/connector/mongodb/ReplicaSetDiscovery.java
View File

@ -13,6 +13,7 @@
import com.mongodb.MongoClient;
import com.mongodb.MongoException;
import com.mongodb.MongoInterruptedException;
import com.mongodb.ReplicaSetStatus;
import io.debezium.annotation.ThreadSafe;
@ -70,6 +71,11 @@ public ReplicaSets getReplicaSets() {
String replicaSetName = MongoUtil.replicaSetUsedIn(hostStr);
replicaSetSpecs.add(new ReplicaSet(hostStr, replicaSetName, shardName));
});
}
catch (MongoInterruptedException e) {
logger.error("Interrupted while reading the '{}' collection in the '{}' database: {}",
shardsCollection, CONFIG_DATABASE_NAME, e.getMessage(), e);
Thread.currentThread().interrupt();
} catch (MongoException e) {
logger.error("Error while reading the '{}' collection in the '{}' database: {}",
shardsCollection, CONFIG_DATABASE_NAME, e.getMessage(), e);

18
debezium-connector-mongodb/src/main/java/io/debezium/connector/mongodb/ReplicaSetMonitorThread.java
View File

@ -13,6 +13,8 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.mongodb.MongoInterruptedException;
import io.debezium.util.Clock;
import io.debezium.util.Metronome;
@ -65,11 +67,21 @@ public void run() {
// At least one of the replica sets been added or been removed ...
try {
onChange.accept(replicaSets);
} catch (Throwable t) {
}
catch (MongoInterruptedException t) {
logger.error("Interrupted while calling the function with the new replica set specifications", t);
Thread.currentThread().interrupt();
}
catch (Throwable t) {
logger.error("Error while calling the function with the new replica set specifications", t);
}
}
} catch (Throwable t) {
}
catch (MongoInterruptedException t) {
logger.error("interrupted while trying to get information about the replica sets", t);
Thread.currentThread().interrupt();
}
catch (Throwable t) {
logger.error("Error while trying to get information about the replica sets", t);
}
// Check again whether we are running before we pause ...
@ -77,7 +89,7 @@ public void run() {
try {
metronome.pause();
} catch (InterruptedException e) {
Thread.interrupted();
Thread.currentThread().interrupt();
}
}
}