Skip to content
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

Jump to bottom

🐛 fix mysql-cdc OOM: use capped queue #4203

Merged
merged 6 commits into from
Jun 23, 2021
Merged

🐛 fix mysql-cdc OOM: use capped queue #4203

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
cadee27
fix mysql-cdc OOM: use capped queue
subodh1810 Jun 18, 2021
6a6014f
Merge branch 'master' into mysql-cdc-oom-fix
subodh1810 Jun 21, 2021
d61bd0e
Merge branch 'master' into mysql-cdc-oom-fix
subodh1810 Jun 21, 2021
7954fa0
add comment
subodh1810 Jun 21, 2021
5c08a6d
add sleep for 10 ms
subodh1810 Jun 22, 2021
a1e5fa7
bump version + only sleep when insertion didnt happen
subodh1810 Jun 23, 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
2 changes: 1 addition & 1 deletion ...ain/resources/config/STANDARD_SOURCE_DEFINITION/435bb9a5-7887-4809-aa58-28c27df0d7ad.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
"sourceDefinitionId": "435bb9a5-7887-4809-aa58-28c27df0d7ad",
"name": "MySQL",
"dockerRepository": "airbyte/source-mysql",
"dockerImageTag": "0.3.7",
"dockerImageTag": "0.3.8",
"documentationUrl": "https://docs.airbyte.io/integrations/sources/mysql",
"icon": "mysql.svg"
}
2 changes: 1 addition & 1 deletion 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 @@ -81,7 +81,7 @@
- sourceDefinitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad
name: MySQL
dockerRepository: airbyte/source-mysql
dockerImageTag: 0.3.7
dockerImageTag: 0.3.8
documentationUrl: https://docs.airbyte.io/integrations/sources/mysql
icon: mysql.svg
- sourceDefinitionId: 2470e835-feaf-4db6-96f3-70fd645acc77
Expand Down
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.3.7
LABEL io.airbyte.version=0.3.8

LABEL io.airbyte.name=airbyte/source-mysql
12 changes: 11 additions & 1 deletion ...rce-mysql/src/main/java/io/airbyte/integrations/source/mysql/DebeziumRecordPublisher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,17 @@ public void start(Queue<ChangeEvent<String, String>> queue) {
// more on the tombstone:
// https://debezium.io/documentation/reference/configuration/event-flattening.html
if (e.value() != null) {
queue.add(e);
boolean inserted = false;
while (!inserted) {
inserted = queue.offer(e);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we do a small sleep here to ensure we don't busy-loop inside this thread?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@sherifnada What do you think would be a right sleep value?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ideally expo backoff for 1/2/4/8/16 ms maybe capping at half a second. But barring that a magic constant (erring on the smaller side) should do. For example 10ms.

if (!inserted) {
try {
Thread.sleep(10);
} catch (InterruptedException interruptedException) {
throw new RuntimeException(interruptedException);
}
}
}
}
})
.using((success, message, error) -> {
Expand Down
7 changes: 6 additions & 1 deletion ...nnectors/source-mysql/src/main/java/io/airbyte/integrations/source/mysql/MySqlSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -233,7 +233,12 @@ public List<AutoCloseableIterator<AirbyteMessage>> getIncrementalIterators(JsonN
final AirbyteFileOffsetBackingStore offsetManager = initializeState(stateManager);
AirbyteSchemaHistoryStorage schemaHistoryManager = initializeDBHistory(stateManager);
FilteredFileDatabaseHistory.setDatabaseName(config.get("database").asText());
final LinkedBlockingQueue<ChangeEvent<String, String>> queue = new LinkedBlockingQueue<>();
/**
* We use 10000 as capacity cause the default queue size and batch size of debezium is :
* {@link io.debezium.config.CommonConnectorConfig#DEFAULT_MAX_BATCH_SIZE} is 2048
* {@link io.debezium.config.CommonConnectorConfig#DEFAULT_MAX_QUEUE_SIZE} is 8192
*/
final LinkedBlockingQueue<ChangeEvent<String, String>> queue = new LinkedBlockingQueue<>(10000);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

any context behind this magic number? (can we add a comment(

final DebeziumRecordPublisher publisher = new DebeziumRecordPublisher(config, catalog, offsetManager, schemaHistoryManager);
publisher.start(queue);

Expand Down