Add e2e ack, checkpointing and metrics to Postgres stream processing #5375
Conversation
oeyh
requested review from
sb2k16,
chenqi0805,
engechas,
san81,
srikanthjg,
graytaylor0,
dinujoh,
kkondaka,
KarstenSchnitter and
dlvenable
as code owners
| @@ -34,4 +34,5 @@ dependencies { | |||
| testImplementation libs.avro.core | |||
| testImplementation libs.parquet.hadoop | |||
| testImplementation libs.parquet.avro | |||
| // testImplementation 'org.slf4j:slf4j-simple:2.0.9' | |||
| @@ -21,6 +23,14 @@ public enum AcknowledgmentStatus { | |||
|
|
|||
| public ChangeEventStatus(final BinlogCoordinate binlogCoordinate, final long timestamp) { | |||
| this.binlogCoordinate = binlogCoordinate; | |||
| this.logSequenceNumber = null; | |||
There was a problem hiding this comment.
Why do we need this constructor ?
There was a problem hiding this comment.
In StreamCheckpointManager, we use a ConcurrentLinkedQueue<ChangeEventStatus> to track change event status. To make this ChangeEventStatus compatible for both MySQL and Postgres, I added logSequenceNumber in it, so MySQL will use binlogCoordinate and Postgres will use logSequenceNumber.
| try { | ||
| eventProcessingTimer.record(() -> function.accept(message)); | ||
| } catch (Exception e) { | ||
| LOG.error("Failed to process change event of type {}", messageType, e); |
There was a problem hiding this comment.
Added metric "replicationLogEntryProcessingErrors" for this failure.
There was a problem hiding this comment.
Is there a data loss here ? It is possible to reprocess this message ?
There was a problem hiding this comment.
I can add retries as well to handle transient issues. But if the message is somehow broken, there could be data loss here.
There was a problem hiding this comment.
Added retries in latest commit.
| } | ||
| } while (!changeEventStatuses.isEmpty()); | ||
| streamCheckpointer.checkpoint(currentChangeEventStatus.getBinlogCoordinate()); | ||
| streamCheckpointer.checkpoint(engineType, currentChangeEventStatus); |
There was a problem hiding this comment.
Add metrics to this class to track checkpointing.
There was a problem hiding this comment.
Added the following metrics:
positiveAcknowledgementSets
negativeAcknowledgementSets
recordsCheckpointed
noDataExtendLeaseCount
giveupPartitionCount
Signed-off-by: Hai Yan <[email protected]>
Signed-off-by: Hai Yan <[email protected]>
Signed-off-by: Hai Yan <[email protected]>
Signed-off-by: Hai Yan <[email protected]>
Signed-off-by: Hai Yan <[email protected]>
oeyh
force-pushed
the
rds/postgres-ack-checkpoint
branch
from
77313d1 to
f6bd2e2
Compare
|
Had to rebase to resolve merge conflicts. |
| changeEventStatus.getBinlogCoordinate() : changeEventStatus.getLogSequenceNumber(), | ||
| changeEventStatus.getRecordCount()); | ||
| streamCheckpointer.checkpoint(engineType, changeEventStatus); | ||
| recordsCheckpointed.increment(changeEventStatus.getRecordCount()); |
There was a problem hiding this comment.
getRecordCount will be total events processed so far. You may want to just increment it by 1 to make sure the checkpoint thread is working as expected.
There was a problem hiding this comment.
Updated the metric to increment by 1 at a time.
Signed-off-by: Hai Yan <[email protected]>
Description
Add e2e ack, checkpointing and metrics to Postgres stream processing.
This is very similar to MySQL stream processing - binlogCoordinate in MySQL vs logSequenceNumber in Postgres. This PR refactors
StreamCheckpointerandStreamCheckpointManagerto support both binlog coordinates and log sequence numbers.Issues Resolved
Contributes to #5309
Check List
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.