DBZ-1174 #806
DBZ-1174 #806
Conversation
- wal2json sends the txn commitTime using a function from PostgreSQL's C library. The value that Debezium recevies is in nanoseconds. - decoderbufs sends the txn commitTime in microseconds. - RecordsSnapshotProducer updates SourceInfo.ts_usec by converting System.currentTimeMillis() to microseconds. - RecordsSnapshotProducer updates the SourceInfo's ts_usec field using message.getCommitTime(). This means that when using wal2json, the value of SourceInfo.ts_usec is in microseconds since epoch during snapshot but is in nanoseconds during streaming. To fix this, we changed Wal2JsonReplicationMessage.getCommitTime() to return in microseconds.
|
@jpechane, can you review and merge this one? Thanks! |
| assertTrue(record.value() instanceof Struct); | ||
| Struct source = ((Struct) record.value()).getStruct("source"); | ||
| // 1 minute difference is okay | ||
| System.out.println("TS_USEC\t" + source.getInt64("ts_usec")); |
There was a problem hiding this comment.
Ooops. I forgot to remove this. Moving this into the assertTrue message would be better.
There was a problem hiding this comment.
Will take care of it.
| // check records from streaming | ||
| inst = Instant.now(); | ||
| // Microseconds since epoch, may overflow | ||
| final long microsStream = TimeUnit.SECONDS.toMicros(inst.getEpochSecond()) + TimeUnit.NANOSECONDS.toMicros(inst.getNano()); |
There was a problem hiding this comment.
Replace with Conversions.toEpochMicros().
| // check records from snapshot | ||
| Instant inst = Instant.now(); | ||
| // Microseconds since epoch, may overflow | ||
| final long microsSnapshot = TimeUnit.SECONDS.toMicros(inst.getEpochSecond()) + TimeUnit.NANOSECONDS.toMicros(inst.getNano()); |
There was a problem hiding this comment.
Replace with Conversions.toEpochMicros().
There was a problem hiding this comment.
Ah yes, good spot. Added that method only later on. I'll update.
| @@ -37,14 +38,17 @@ static Clock system() { | |||
| return SYSTEM; | |||
| } | |||
|
|
|||
| default Instant currentTime() { | |||
| return Instant.ofEpochMilli(currentTimeInMillis()); | |||
There was a problem hiding this comment.
Any reason to not use Instant.now()? Is that to allow for mocking the Clock in tests?
There was a problem hiding this comment.
Yes, the idea is to make Clock implementations pluggable (and their only required method is currentTimeInMillis().
|
Thanks for reviewing, @hashhar! I'll push an update in a bit. |
| @@ -53,7 +55,7 @@ public void processMessage(ByteBuffer buffer, ReplicationMessageProcessor proces | |||
| LOGGER.debug("Message arrived for decoding {}", message); | |||
| final long txId = message.getLong("xid"); | |||
| final String timestamp = message.getString("timestamp"); | |||
| final long commitTime = dateTime.systemTimestamp(timestamp); | |||
| final Instant commitTime = Conversions.toInstant(dateTime.systemTimestamp(timestamp)); | |||
There was a problem hiding this comment.
@gunnarmorling Should not systemTimestamp return Instant instead of doing conversion here?
There was a problem hiding this comment.
Ideally yes. But I was trying, this is like pulling Sphaghetti from the plate, never ending. So I decided to stop here, and leave that for another time.
|
@hashhar @gunnarmorling Thanks guys! |
https://issues.jboss.org/browse/DBZ-1174