alexey.rassokhin/tet123
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

DBZ-1840 Typo fix; commenting

Browse Source
This commit is contained in:
Gunnar Morling 2020-03-04 09:31:22 +01:00
parent 10dfc12c99
commit a8b796685b
2 changed files with 10 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/connection/ReplicationMessage.java
View File

@ -245,17 +245,16 @@ public Instant getCommitTime() {
/**
* A special message type that is used to replace event filtered already at {@link MessageDecoder}.
* Enables {@link PostgresStreamingChangeEventSource} to advance LSN forward even in case of such messages.
*
*/
public class NoopMessage implements ReplicationMessage {
private final long transationId;
private final long transactionId;
private final Instant commitTime;
private final Operation operation;
public NoopMessage(long transactionId, Instant commitTime) {
this.operation = Operation.NOOP;
this.transationId = transactionId;
this.transactionId = transactionId;
this.commitTime = commitTime;
}
@ -271,7 +270,7 @@ public boolean hasTypeMetadata() {
@Override
public long getTransactionId() {
return transationId;
return transactionId;
}
@Override
@ -298,5 +297,5 @@ public List<Column> getNewTupleList() {
public Instant getCommitTime() {
return commitTime;
}
};
}
}

6
debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/connection/pgoutput/PgOutputMessageDecoder.java
View File

@ -364,6 +364,8 @@ private void decodeInsert(ByteBuffer buffer, TypeRegistry typeRegistry, Replicat
LOGGER.trace("Event: {}, Relation Id: {}, Tuple Type: {}", MessageType.INSERT, relationId, tupleType);
Optional<Table> resolvedTable = resolveRelation(relationId);
// non-captured table
if (!resolvedTable.isPresent()) {
processor.process(new NoopMessage(transactionId, commitTimestamp));
}
@ -393,6 +395,8 @@ private void decodeUpdate(ByteBuffer buffer, TypeRegistry typeRegistry, Replicat
LOGGER.trace("Event: {}, RelationId: {}", MessageType.UPDATE, relationId);
Optional<Table> resolvedTable = resolveRelation(relationId);
// non-captured table
if (!resolvedTable.isPresent()) {
processor.process(new NoopMessage(transactionId, commitTimestamp));
}
@ -440,6 +444,8 @@ private void decodeDelete(ByteBuffer buffer, TypeRegistry typeRegistry, Replicat
LOGGER.trace("Event: {}, RelationId: {}, Tuple Type: {}", MessageType.DELETE, relationId, tupleType);
Optional<Table> resolvedTable = resolveRelation(relationId);
// non-captured table
if (!resolvedTable.isPresent()) {
processor.process(new NoopMessage(transactionId, commitTimestamp));
}