SourceInfo extends BaseSourceInfo { public Map<String, String> partition(String replicaSetName) { if (replicaSetName == null) { throw new IllegalArgumentException("Replica set name may not be null"); } return sourcePartitionsByReplicaSetName.computeIfAbsent(replicaSetName, rsName -> { return Collect.hashMapOf(SERVER_ID_KEY, serverName(), REPLICA_SET_NAME, rsName); }); } 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 shouldProducePartitionMap() { partition = source.partition(REPLICA_SET_NAME); assertThat(partition.get(SourceInfo.REPLICA_SET_NAME)).isEqualTo(REPLICA_SET_NAME); assertThat(partition.get(SourceInfo.SERVER_ID_KEY)).isEqualTo("serverX"); assertThat(partition.size()).isEqualTo(2); }
@Test public void shouldReturnSamePartitionMapForSameReplicaName() { partition = source.partition(REPLICA_SET_NAME); assertThat(partition).isSameAs(source.partition(REPLICA_SET_NAME)); }