https://linen.dev logo
Join Slack
Powered by
If i have a connection to a postgres db that has a...
# replication-troubleshooting
r
If i have a connection to a postgres db that has a max 15min timeout that writes to snowflake, it can't finish the full sync in that amount of, is there a way for the sync to resume from where it stopped? atleast start from any tables that did not finish vs restarting all of them each time? The goal would be to run it a few times with failures until all the data is sync, then the incremental sync would be much shorter going forward once the initial is done
✍️ 1
u
@[DEPRECATED] Marcos Marx turned this message into Zendesk ticket 2868 to ensure timely resolution!
m
I opened an issue for that, but seems like it wasn't prioritized yet, if more people ask for it maybe Airbyte team speed that up. https://github.com/airbytehq/airbyte/issues/17774
r
so is there a workaround if i keep running the sync?
let it fail 3 times
and then again until it all syncs?
m
Yes, fix attempts in 1
A second job will continue from where it stopped
s
Hey @Robert Put, Unfortunately, this is also still a feature we have on our roadmap. There's the issue that @Marcelo Pio de Castro made and there's an issue that our team made tracking this issue: https://github.com/airbytehq/airbyte/issues/16762 The issue at the moment is that we don't persist state during syncs, so other parts of our platform are not aware if work is performed already, so the only signal for redoing work is if a sync fails. There have been some internal discussions on when we plan to roll this out. But I don't have a clear timeline for you atm
m
Hey @Saj Dider (Airbyte) , I think my issue is a lot easier to fix as we do persist the state between attempts, Airbyte just don't use it.
s
@Marcelo Pio de Castro, it might be easier to fix. I haven't replicated the bug for myself so I'm hesitant to conclude anything.
m
That is ok. If you need any help replicating or just want for me to show in a call, I am available, as that issue is critical for me. Thanks!
r
1 more question around options for getting this to work without needing to make any db config changes. Can i add 1 table to replica, let it finish, then add another, continuing until all are managed, as long as each can be replicated in the needed time. Or does adding a new table stream cause all stream to reset?
m
You don't need to reset the stream if you don't want to, just click cancel on the window that opens. As for the attempt config, you just need to config a variable in the .env airbyte file, SYNC_JOB_MAX_ATTEMPTS, put a value of 1 in it. That way it will only have one attempt, but the next sync will continue from where the the job before failed
🙏 1
r
let me try, thank you!
@Marcelo Pio de Castro to verify, there is no data lose or gaps between the attemps? im running incremental dedup syncs
m
Hmm, attempts will try the whole sync again, so it kinda of sends more data that already was sync. If you set only one attempt, you won't lose data, it will just continue from where the latest state was emitted, it can still duplicate data (airbyte guarantees at least once delivery) but you won't lose data
r
and then debup would resolve any issues in that case?
m
Yes, dedup will remove the duplicates
r
perfect, let me give this a try
what happens if my ec2 instance crashed, but im running remote postgres? it just never sees the sync state as having been successful so it restarts form the last time it saw data get to the destination successfully?
wondering about the local staging of data before it writes to the destination
m
It will emit state messages and save the last one, so it will have a state of the last cursor field that it saved. It saves the cursor periodically, so it will restart from that onwards, you config the cursor field on the connection. You can check what is the state of the cursor field in the connection if you enable advanced mode in Configuration -> Account
Then go to the connection, you can see the state for each table in the connection settings
r
i kept toggling that and couldn't see what was changing, thanks for that info
m
It won't save the state during the connection, that is the issue Saj mentioned, so if the Airbyte instance crashes, it will restart from where the state was saved, probably from the latest sync that finished, even if it failed or was successful
r
this sounds perfect, thanks for all the info. running my tests, hopefully this is the last hurdle before I can adopt it over stitch
it seems to be trying to replicate the same tables each time it tries again? getting about as far each time?
getting emitted records but nothing to the destination
seems like ~4 streams finish, it has 30 configured, on the 5th it crashes, shows x records emitted, and then when i run sync now again, it starts from the start?
m
Strange... Can you check if the states were saved?
On the connection -> Settings
r
no
as in, not saved, its empty
i see this in the logs:
Copy code
2022-10-17 18:12:42 WARN i.a.w.g.DefaultReplicationWorker(run):289 - State capture: No new state, falling back on input state: io.airbyte.config.State@1017059e[state={}]
does it need some initial run to fall back to?
Copy code
2022-10-17 18:12:42 ERROR i.a.w.g.DefaultReplicationWorker(run):188 - Sync worker failed.
java.util.concurrent.ExecutionException: io.airbyte.workers.general.DefaultReplicationWorker$SourceException: Source cannot be stopped!
	at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:396) ~[?:?]
	at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2073) ~[?:?]
	at io.airbyte.workers.general.DefaultReplicationWorker.run(DefaultReplicationWorker.java:181) ~[io.airbyte-airbyte-workers-0.40.14.jar:?]
	at io.airbyte.workers.general.DefaultReplicationWorker.run(DefaultReplicationWorker.java:68) ~[io.airbyte-airbyte-workers-0.40.14.jar:?]
	at io.airbyte.workers.temporal.TemporalAttemptExecution.lambda$getWorkerThread$2(TemporalAttemptExecution.java:161) ~[io.airbyte-airbyte-workers-0.40.14.jar:?]
	at java.lang.Thread.run(Thread.java:833) ~[?:?]
	Suppressed: io.airbyte.workers.exception.WorkerException: Source process exit with code 1. This warning is normal if the job was cancelled.
		at io.airbyte.workers.internal.DefaultAirbyteSource.close(DefaultAirbyteSource.java:135) ~[io.airbyte-airbyte-workers-0.40.14.jar:?]
		at io.airbyte.workers.general.DefaultReplicationWorker.run(DefaultReplicationWorker.java:145) ~[io.airbyte-airbyte-workers-0.40.14.jar:?]
		at io.airbyte.workers.general.DefaultReplicationWorker.run(DefaultReplicationWorker.java:68) ~[io.airbyte-airbyte-workers-0.40.14.jar:?]
		at io.airbyte.workers.temporal.TemporalAttemptExecution.lambda$getWorkerThread$2(TemporalAttemptExecution.java:161) ~[io.airbyte-airbyte-workers-0.40.14.jar:?]
		at java.lang.Thread.run(Thread.java:833) ~[?:?]
	Suppressed: io.airbyte.workers.exception.WorkerException: Destination process exit with code 143. This warning is normal if the job was cancelled.
		at io.airbyte.workers.internal.DefaultAirbyteDestination.close(DefaultAirbyteDestination.java:115) ~[io.airbyte-airbyte-workers-0.40.14.jar:?]
		at io.airbyte.workers.general.DefaultReplicationWorker.run(DefaultReplicationWorker.java:145) ~[io.airbyte-airbyte-workers-0.40.14.jar:?]
		at io.airbyte.workers.general.DefaultReplicationWorker.run(DefaultReplicationWorker.java:68) ~[io.airbyte-airbyte-workers-0.40.14.jar:?]
seems related?
right before that message it tries to write out to the destination:
Copy code
2022-10-17 18:11:55 destination > Copying stream 1
2022-10-17 18:11:55 destination > Copying stream 2
2022-10-17 18:11:55 destination > Copying stream 3
2022-10-17 18:11:55 destination > Copying stream 4
2022-10-17 18:12:42 ERROR i.a.w.g.DefaultReplicationWorker(run):188 - Sync worker failed.
is there something else i need to set to get it to work as described?
Copy code
### JOBS ###
# Relevant to scaling.
SYNC_JOB_MAX_ATTEMPTS=1
SYNC_JOB_MAX_TIMEOUT_DAYS=3
JOB_MAIN_CONTAINER_CPU_REQUEST=
JOB_MAIN_CONTAINER_CPU_LIMIT=
JOB_MAIN_CONTAINER_MEMORY_REQUEST=
JOB_MAIN_CONTAINER_MEMORY_LIMIT=

NORMALIZATION_JOB_MAIN_CONTAINER_MEMORY_LIMIT=
NORMALIZATION_JOB_MAIN_CONTAINER_MEMORY_REQUEST=
NORMALIZATION_JOB_MAIN_CONTAINER_CPU_LIMIT=
NORMALIZATION_JOB_MAIN_CONTAINER_CPU_REQUEST=

### LOGGING/MONITORING/TRACKING ###
TRACKING_STRATEGY=logging
PAPERCUPS_STORYTIME=disabled
JOB_ERROR_REPORTING_STRATEGY=logging
# Although not present as an env var, expected by Log4J configuration.
LOG_LEVEL=INFO


### APPLICATIONS ###
# Worker #
WORKERS_MICRONAUT_ENVIRONMENTS=control-plane
# Relevant to scaling.
MAX_SYNC_WORKERS=5
MAX_SPEC_WORKERS=5
MAX_CHECK_WORKERS=5
MAX_DISCOVER_WORKERS=5
m
The only thing you need to enable for Airbyte to send state messages is that your sync must be incremental. Full refresh do not emit or save state messages.
Even if the source fails, you should see something like this:
Copy code
2022-10-09 23:11:20 [32mINFO[m i.a.w.g.DefaultReplicationWorker(run):279 - Source output at least one state message
2022-10-09 23:11:20 [32mINFO[m i.a.w.g.DefaultReplicationWorker(run):285 - State capture: Updated state to: Optional[io.airbyte.config.State@1d4b5b2f[state=[{"type":"STREAM","stream":{"stream_descriptor":{"name":"Test"},"stream_state":{"last_number":4}}}]]]
r
there are running all as incremental debup
is the issue that the first sync is a full refresh followed by incremental after?
m
You can't configure the first sync as a full refresh, it needs to be incremental since the start. As the state won't exist in the first sync, it will basically do a full refresh, but the state will be emited
r
im just not seeing any state changes when doing it
Screen Shot 2022-10-17 at 3.40.04 PM.png
its does the same tables and gets just as far each time:
m
Hmm, in the logged queries in the job do you see an 
ORDER BY <cursor_field> ASC
?
r
nope, 
ORDER BY
doesn't even appear in the logs 
updated_at asc
doesn't appear either
m
But you do see this right? 
Prepared SQL query for queryTableIncremental is
r
not that either
i was thinking that the initial sync is the issue due to:
Copy code
If the sync is running for the first time, it is equivalent to a Full Refresh since all data will be considered as new.
also using latest airbyte, postgres source and snowflake destinations
m
I am investigating this PR, may be related... https://github.com/airbytehq/airbyte/pull/15496
r
so it sounds like its working as expected but not as desired? since there are more records with those cursor values its not pushing state untill they all get pushed which would be everything the first time?
m
I really don't know for now, just investigating still, from the code, it should have a 
ORDER BY
if emitting states periodically https://github.com/airbytehq/airbyte/blob/a5a0409ad55192abda4b6fac028ac90080e51e43[…]ava/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java
Don't know why that is not working
r
m
Did you enable per stream state?
r
Copy code
USE_STREAM_CAPABLE_STATE
?
Did you enable per stream state?
what does that mean?
but not sure if there is a value to toggle or check anywhere?
I did push all these init values using octavia also, not sure if that impacts it
m
USE_STREAM_CAPABLE_STATE
, yeah, that is the flag
r
yeah its set to true
Copy code
2022-10-17 19:35:15 [32mINFO[m i.a.w.p.DockerProcessFactory(create):163 - Preparing command: docker run --rm --init -i -w /data/48/0 --log-driver none --name source-postgres-read-48-0-urmmk --network host -v airbyte_workspace:/data -v /tmp/airbyte_local:/local -e DEPLOYMENT_MODE=OSS -e USE_STREAM_CAPABLE_STATE=true -e WORKER_ENVIRONMENT=DOCKER -e AIRBYTE_ROLE= -e WORKER_JOB_ATTEMPT=0 -e WORKER_CONNECTOR_IMAGE=airbyte/source-postgres:1.0.16 -e AIRBYTE_VERSION=0.40.14 -e WORKER_JOB_ID=48 airbyte/source-postgres:1.0.16 read --config source_config.json --catalog source_catalog.json --state input_state.json
and yeah its the default 
true
in the env file
m
r
is that configured in the connection or in the source? since this was pushed using the cli would it be a missing flag?
m
It is configured in the source. I found the issue, your hipothesis was correct, since this is the first sync, Airbyte it not emiting intemediate states, it only emits an state at the end of the stream. I would classify that as an issue, I don't think that is the way it should be
r
wait
nvm it fixing it for oracle but was part of the psotgres source versioning?
but in my case, some of the streams are finishing but not emitting any states, does it also have to be all streams?
m
Do you have a state report in your logs? 
State Report: stream name
r
Copy code
2022-10-17 19:35:19 [44msource[0m > State report for stream AirbyteStreamNameNamespacePair{name='app_config', namespace='public'} - original: null = null (count 0) -> latest: updated_at = 2022-10-13T10:09:56.845870 (count 1)
yeah a ton
Copy code
2022-10-17 19:35:42 [44msource[0m > State report for stream AirbyteStreamNameNamespacePair{name='test', namespace='public'} - original: null = null (count 0) -> latest: created_at = 2022-10-17T15:57:49.000000 (count 1)
nvm
Copy code
State report:
doesn't appear
m
mine state report might be different, but those are the states being emitted, those should be persisted to Airbyte....
You are probably getting this, correct? 
Source did not output any state messages
r
Copy code
2022-10-17 20:09:41 INFO i.a.w.g.DefaultReplicationWorker(run):279 - Source output at least one state message
2022-10-17 20:09:41 WARN i.a.w.g.DefaultReplicationWorker(run):289 - State capture: No new state, falling back on input state: io.airbyte.config.State@69c7fbb0[state={}]
Copy code
2022-10-17 20:09:41 ERROR i.a.w.g.DefaultReplicationWorker(run):188 - Sync worker failed.
java.util.concurrent.ExecutionException: io.airbyte.workers.general.DefaultReplicationWorker$SourceException: Source cannot be stopped!
wait maybe its an oom issue with the sync worker?
Copy code
io.airbyte.config.StreamSyncStats@4b3043e5[streamName=app_config,stats=io.airbyte.config.SyncStats@14e0d688[recordsEmitted=1,bytesEmitted=1463,sourceStateMessagesEmitted=<null>
Copy code
ourceStateMessagesEmitted=<null>
?
m
This is probably happening because Snowflake didn't actually flush any records
I think I saw an issue for it, let me get it
r
thats related to the internal staging?
m
Since your records don't actually make it to the final _raw table, it does not emit the state (which is correct)
So, problem is that 14770 is not implemented yet...
r
yeah its all in the temp table
😪
m
Yeah... if it was from postgres to postgres it would work..., snowflake, and in my case, BigQuery won't, that is an issue for me also
r
for this db i can push the timeout to like ~4 hours, and get all the data in.... nothing else uses the read replica.... put my other dbs are larger and used hourly for live data currently
i guess i spin up a temp read replica just for me for the initial extraction?
m
might work, just change the source host
the connection won't even notice
r
any better ideas?
appreciate all the help!
Open in Slack
PreviousNext
Powered by