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

DBZ-1723 Encapsulating common poll() and stop() logic in BaseSourceTask

Browse Source
This commit is contained in:
Gunnar Morling 2020-03-02 11:25:15 +01:00 committed by Jiri Pechanec
parent 6318cf0f2b
commit 8ce69dc597
5 changed files with 15 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
debezium-connector-mongodb/src/main/java/io/debezium/connector/mongodb/MongoDbConnectorTask.java
View File

@ -124,7 +124,7 @@ public ChangeEventSourceCoordinator start(Configuration config) {
}
@Override
public List<SourceRecord> poll() throws InterruptedException {
public List<SourceRecord> doPoll() throws InterruptedException {
List<DataChangeEvent> records = queue.poll();
return records.stream().map(DataChangeEvent::getRecord).collect(Collectors.toList());
}

2
debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/MySqlConnectorTask.java
View File

@ -423,7 +423,7 @@ private static Filters getAllFilters(Configuration config) {
}
@Override
public List<SourceRecord> poll() throws InterruptedException {
public List<SourceRecord> doPoll() throws InterruptedException {
Reader currentReader = readers;
if (currentReader == null) {
return null;

2
debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresConnectorTask.java
View File

@ -200,7 +200,7 @@ public ReplicationConnection createReplicationConnection(PostgresTaskContext tas
}
@Override
public List<SourceRecord> poll() throws InterruptedException {
public List<SourceRecord> doPoll() throws InterruptedException {
final List<DataChangeEvent> records = queue.poll();
final List<SourceRecord> sourceRecords = records.stream()

2
debezium-connector-sqlserver/src/main/java/io/debezium/connector/sqlserver/SqlServerConnectorTask.java
View File

@ -126,7 +126,7 @@ public ChangeEventSourceCoordinator start(Configuration config) {
}
@Override
public List<SourceRecord> poll() throws InterruptedException {
public List<SourceRecord> doPoll() throws InterruptedException {
final List<DataChangeEvent> records = queue.poll();
final List<SourceRecord> sourceRecords = records.stream()

11
debezium-core/src/main/java/io/debezium/connector/common/BaseSourceTask.java
View File

@ -6,6 +6,7 @@
package io.debezium.connector.common;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
@ -86,6 +87,16 @@ public final void start(Map<String, String> props) {
*/
protected abstract ChangeEventSourceCoordinator start(Configuration config);
@Override
public final List<SourceRecord> poll() throws InterruptedException {
return doPoll();
}
/**
* Returns the next batch of source records, if any are available.
*/
public abstract List<SourceRecord> doPoll() throws InterruptedException;
@Override
public void commitRecord(SourceRecord record) throws InterruptedException {
Map<String, ?> currentOffset = record.sourceOffset();