SourceInfo extends BaseSourceInfo { public void opLogEvent(String replicaSetName, Document oplogEvent, Document masterEvent, long orderInTx) { Position position = INITIAL_POSITION; String namespace = ""; if (oplogEvent != null) { BsonTimestamp ts = extractEventTimestamp(masterEvent); Long opId = masterEvent.getLong("h"); String sessionTxnId = extractSessionTxnId(masterEvent); position = new Position(ts, opId, orderInTx, sessionTxnId); namespace = oplogEvent.getString("ns"); } positionsByReplicaSetName.put(replicaSetName, position); onEvent(replicaSetName, CollectionId.parse(replicaSetName, namespace), position); } SourceInfo(MongoDbConnectorConfig connectorConfig); static String replicaSetNameForPartition(Map<String, ?> partition); Map<String, String> partition(String replicaSetName); BsonTimestamp lastOffsetTimestamp(String replicaSetName); OptionalLong lastOffsetTxOrder(String replicaSetName); Map<String, ?> lastOffset(String replicaSetName); void collectionEvent(String replicaSetName, CollectionId collectionId); void opLogEvent(String replicaSetName, Document oplogEvent, Document masterEvent, long orderInTx); void opLogEvent(String replicaSetName, Document oplogEvent); boolean hasOffset(String replicaSetName); boolean setOffsetFor(String replicaSetName, Map<String, ?> sourceOffset); boolean setOffsetFor(Map<String, String> partition, Map<String, ?> sourceOffset); void startInitialSync(String replicaSetName); void stopInitialSync(String replicaSetName); boolean isInitialSyncOngoing(String replicaSetName); boolean isSnapshotRunning(); static final int SCHEMA_VERSION; static final String SERVER_ID_KEY; static final String REPLICA_SET_NAME; static final String NAMESPACE; static final String TIMESTAMP; static final String ORDER; static final String OPERATION_ID; static final String TX_ORD; static final String SESSION_TXN_ID; static final String INITIAL_SYNC; static final String COLLECTION; }
@Test public void versionIsPresent() { final Document event = new Document().append("ts", new BsonTimestamp(100, 2)) .append("h", Long.valueOf(1987654321)) .append("ns", "dbA.collectA"); source.opLogEvent("rs", event); assertThat(source.struct().getString(SourceInfo.DEBEZIUM_VERSION_KEY)).isEqualTo(Module.version()); }
@Test public void connectorIsPresent() { final Document event = new Document().append("ts", new BsonTimestamp(100, 2)) .append("h", Long.valueOf(1987654321)) .append("ns", "dbA.collectA"); source.opLogEvent("rs", event); assertThat(source.struct().getString(SourceInfo.DEBEZIUM_CONNECTOR_KEY)).isEqualTo(Module.name()); }