I tried the

batch_wait_limit_seconds: 200

option. However, if table A has finished, but not flushed, and table B is currently syncing, then table B will be flushed 200 seconds after table A batch started, which is not at all what I want!

To your main question though, I agree that it would likely be preferrable to only flush those streams that have records older than the

batch_wait_limit_seconds

. Two possible improvements would be: 1. Replace

flush_all()

with something more stream-specific, so that in your example "Stream A" would load but not "Stream b". 2. Add an additional arg to

flush_all()

such as

flush_all(min_record_age=batch_wait_limit_seconds)

which would operate conditionally to flush only those streams that have records over the specified threshold. 3. Same to the above but flush anything with a record age that's nearing the 75% point of the max wait limit. This prevents a case where the records of Stream B are aging as Stream A gets synced and then as soon as Stream A is done we end up having to drain Stream B anyway, and the more ragged approach may be a less optimal load pattern than draining both at once. What do you think of these approaches?

For context, we used to do a sync/drain after every state message. This introduced a HUGE slowdown for some taps that were sending multiple state messages per second. 🙂 What we have now is a large step better than the prior algorithm, but still has room for improvement - for sure.

and the current behavior is making some errors difficult to recover from

For this target (the one from Pipelinewise), it looks like the logic is such that the logic is supposed to flush each table (individually) if either the max records are reached or the wait limit seconds have passed.

flush = False
            if row_count[stream] >= batch_size_rows:
                flush = True
                <http://LOGGER.info|LOGGER.info>("Flush triggered by batch_size_rows (%s) reached in %s",
                            batch_size_rows, stream)
            elif (batch_wait_limit_seconds and
                  datetime.utcnow() >= (flush_timestamp + timedelta(seconds=batch_wait_limit_seconds))):
                flush = True
                <http://LOGGER.info|LOGGER.info>("Flush triggered by batch_wait_limit_seconds (%s)",
                            batch_wait_limit_seconds)

Sorry for leading you the wrong direction on this. As of now, for that target, I think those are the only to levers to control when the records are flushed.