Skip to content

remove sleep logic when the queue is full from CDC #5600

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Aug 30, 2021
Merged

remove sleep logic when the queue is full from CDC #5600

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
345210f
dont sleep when queue is full
subodh1810 Aug 24, 2021
9f36df1
Merge branch 'master' into fix-interrupted-exception
subodh1810 Aug 25, 2021
f262df7
bump version
subodh1810 Aug 26, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions airbyte-config/init/src/main/resources/seed/source_definitions.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@
- sourceDefinitionId: b5ea17b1-f170-46dc-bc31-cc744ca984c1
name: Microsoft SQL Server (MSSQL)
dockerRepository: airbyte/source-mssql
dockerImageTag: 0.3.4
dockerImageTag: 0.3.5
documentationUrl: https://docs.airbyte.io/integrations/sources/mssql
icon: mssql.svg
- sourceDefinitionId: d8286229-c680-4063-8c59-23b9b391c700
Expand All @@ -56,7 +56,7 @@
- sourceDefinitionId: decd338e-5647-4c0b-adf4-da0e75f5a750
name: Postgres
dockerRepository: airbyte/source-postgres
dockerImageTag: 0.3.9
dockerImageTag: 0.3.10
documentationUrl: https://docs.airbyte.io/integrations/sources/postgres
icon: postgresql.svg
- sourceDefinitionId: 9fa5862c-da7c-11eb-8d19-0242ac130003
Expand Down Expand Up @@ -101,7 +101,7 @@
- sourceDefinitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad
name: MySQL
dockerRepository: airbyte/source-mysql
dockerImageTag: 0.4.3
dockerImageTag: 0.4.4
documentationUrl: https://docs.airbyte.io/integrations/sources/mysql
icon: mysql.svg
- sourceDefinitionId: 2470e835-feaf-4db6-96f3-70fd645acc77
Expand Down
8 changes: 5 additions & 3 deletions ...zium/src/main/java/io/airbyte/integrations/debezium/internals/DebeziumRecordIterator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ public class DebeziumRecordIterator extends AbstractIterator<ChangeEvent<String,
private final VoidCallable requestClose;
private boolean receivedFirstRecord;
private boolean hasSnapshotFinished;
private boolean signalledClose;

public DebeziumRecordIterator(LinkedBlockingQueue<ChangeEvent<String, String>> queue,
CdcTargetPosition targetPosition,
Expand All @@ -74,6 +75,7 @@ public DebeziumRecordIterator(LinkedBlockingQueue<ChangeEvent<String, String>> q
this.requestClose = requestClose;
this.receivedFirstRecord = false;
this.hasSnapshotFinished = true;
this.signalledClose = false;
}

@Override
Expand Down Expand Up @@ -103,7 +105,7 @@ protected ChangeEvent<String, String> computeNext() {
hasSnapshotFinished = hasSnapshotFinished(eventAsJson);

// if the last record matches the target file position, it is time to tell the producer to shutdown.
if (shouldSignalClose(eventAsJson)) {
if (!signalledClose && shouldSignalClose(eventAsJson)) {
requestClose();
}
receivedFirstRecord = true;
Expand Down Expand Up @@ -135,8 +137,7 @@ private boolean hasSnapshotFinished(JsonNode eventAsJson) {
*/
@Override
public void close() throws Exception {
requestClose.call();
throwExceptionIfSnapshotNotFinished();
requestClose();
}

private boolean shouldSignalClose(JsonNode eventAsJson) {
Expand All @@ -146,6 +147,7 @@ private boolean shouldSignalClose(JsonNode eventAsJson) {
private void requestClose() {
try {
requestClose.call();
signalledClose = true;
} catch (Exception e) {
throw new RuntimeException(e);
}
Expand Down
7 changes: 0 additions & 7 deletions ...ium/src/main/java/io/airbyte/integrations/debezium/internals/DebeziumRecordPublisher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,13 +98,6 @@ public void start(Queue<ChangeEvent<String, String>> queue) {
boolean inserted = false;
while (!inserted) {
inserted = queue.offer(e);
if (!inserted) {
try {
Thread.sleep(10);
} catch (InterruptedException interruptedException) {
throw new RuntimeException(interruptedException);
}
}
}
}
})
Expand Down
2 changes: 1 addition & 1 deletion airbyte-integrations/connectors/source-mssql/Dockerfile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar

RUN tar xf ${APPLICATION}.tar --strip-components=1

LABEL io.airbyte.version=0.3.4
LABEL io.airbyte.version=0.3.5
LABEL io.airbyte.name=airbyte/source-mssql
2 changes: 1 addition & 1 deletion airbyte-integrations/connectors/source-mysql/Dockerfile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,6 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar

RUN tar xf ${APPLICATION}.tar --strip-components=1

LABEL io.airbyte.version=0.4.3
LABEL io.airbyte.version=0.4.4

LABEL io.airbyte.name=airbyte/source-mysql
2 changes: 1 addition & 1 deletion airbyte-integrations/connectors/source-postgres/Dockerfile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar

RUN tar xf ${APPLICATION}.tar --strip-components=1

LABEL io.airbyte.version=0.3.9
LABEL io.airbyte.version=0.3.10
LABEL io.airbyte.name=airbyte/source-postgres