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

DBZ-7759 Use MongoException in error handling

Browse Source
This commit is contained in:
ani-sha 2024-04-11 12:49:09 +05:30 committed by Jiri Pechanec
parent 1b7481d357
commit aef55a1ccd
1 changed files with 8 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
debezium-connector-mongodb/src/main/java/io/debezium/connector/mongodb/events/BufferingChangeStreamCursor.java
View File

@ -22,7 +22,6 @@
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import com.mongodb.MongoChangeStreamException;
import com.mongodb.MongoException; import com.mongodb.MongoException;
import com.mongodb.ServerAddress; import com.mongodb.ServerAddress;
import com.mongodb.ServerCursor; import com.mongodb.ServerCursor;
@ -227,13 +226,13 @@ private void fetchEvents(MongoChangeStreamCursor<ChangeStreamDocument<TResult>>
private Optional<ResumableChangeStreamEvent<TResult>> fetchEvent(MongoChangeStreamCursor<ChangeStreamDocument<TResult>> cursor) { private Optional<ResumableChangeStreamEvent<TResult>> fetchEvent(MongoChangeStreamCursor<ChangeStreamDocument<TResult>> cursor) {
var beforeEventPollTime = clock.currentTimeAsInstant(); var beforeEventPollTime = clock.currentTimeAsInstant();
ChangeStreamDocument<TResult> document; ChangeStreamDocument<TResult> document = null;
try { try {
document = cursor.tryNext(); document = cursor.tryNext();
} }
catch (MongoException e) { catch (MongoException e) {
running.set(false); running.set(false);
throw new MongoChangeStreamException("Error while fetching change stream event" + e.getMessage()); LOGGER.error("Error while fetching change stream event", e);
} }
metrics.onSourceEventPolled(document, clock, beforeEventPollTime); metrics.onSourceEventPolled(document, clock, beforeEventPollTime);
throttleIfNeeded(document); throttleIfNeeded(document);
@ -307,18 +306,13 @@ public BufferingChangeStreamCursor<TResult> start() {
@Override @Override
public ResumableChangeStreamEvent<TResult> tryNext() { public ResumableChangeStreamEvent<TResult> tryNext() {
var event = pollWithDelay(); var event = pollWithDelay();
try { if (event != null) {
if (event != null) { lastResumeToken = event.resumeToken;
lastResumeToken = event.resumeToken;
}
else {
if (!fetcher.isRunning() && fetcher.isEmpty()) {
throw new MongoChangeStreamException("Failed to fetch event from change stream cursor, cursor is closed");
}
}
} }
catch (MongoException e) { else {
throw new MongoException(e.getMessage()); if (!fetcher.isRunning() && fetcher.isEmpty()) {
throw new MongoException("Fetcher thread has stopped and buffer is empty");
}
} }
return event; return event;
} }