Package org.apache.flink.cdc.connectors.mongodb.source.utils

Class MongoUtils

java.lang.Object

org.apache.flink.cdc.connectors.mongodb.source.utils.MongoUtils

public class MongoUtils
extends Object

Utilities of MongoDB operations.

Field Summary

Fields

Modifier and Type	Field	Description
static int	BSON_OBJECT_TOO_LARGE
static int	CHANGE_STREAM_FATAL_ERROR
static int	CHANGE_STREAM_HISTORY_LOST
static int	CLOSE_CHANGE_STREAM
static org.bson.BsonDouble	COMMAND_SUCCEED_FLAG
static int	FAILED_TO_PARSE_ERROR
static int	ILLEGAL_OPERATION_ERROR
static int	INVALIDATED_RESUME_TOKEN_ERROR
static int	RETRY_CHANGE_STREAM
static int	UNAUTHORIZED_ERROR
static int	UNKNOWN_FIELD_ERROR

Method Summary

Modifier and Type	Method	Description
static String	buildConnectionString(String username, String password, String scheme, String hosts, String connectionOptions)
static boolean	checkIfChangeStreamCursorExpires(com.mongodb.MongoCommandException e)
static boolean	checkIfResumeTokenExpires(com.mongodb.MongoCommandException e)
static com.mongodb.client.MongoClient	clientFor(MongoDBSourceConfig sourceConfig)
static <T> com.mongodb.client.MongoCollection<T>	collectionFor(com.mongodb.client.MongoClient mongoClient, io.debezium.relational.TableId collectionId, Class<T> documentClass)
static org.bson.BsonDocument	collStats(com.mongodb.client.MongoClient mongoClient, io.debezium.relational.TableId collectionId)
static String	commandErrorMessage(org.bson.BsonDocument commandResult)
static ChangeStreamDescriptor	getChangeStreamDescriptor(MongoDBSourceConfig sourceConfig, List<String> discoveredDatabases, List<String> discoveredCollections)
static com.mongodb.client.ChangeStreamIterable<org.bson.Document>	getChangeStreamIterable(com.mongodb.client.MongoClient mongoClient, String database, String collection, Pattern databaseRegex, Pattern namespaceRegex, int batchSize, boolean updateLookup, boolean fullDocPrePostImage)
static com.mongodb.client.ChangeStreamIterable<org.bson.Document>	getChangeStreamIterable(com.mongodb.client.MongoClient mongoClient, ChangeStreamDescriptor descriptor, int batchSize, boolean updateLookup, boolean fullDocPrePostImage)
static com.mongodb.client.ChangeStreamIterable<org.bson.Document>	getChangeStreamIterable(MongoDBSourceConfig sourceConfig, ChangeStreamDescriptor descriptor)
static org.bson.BsonTimestamp	getCurrentClusterTime(com.mongodb.client.MongoClient mongoClient)
static org.bson.BsonDocument	getLatestResumeToken(com.mongodb.client.MongoClient mongoClient, ChangeStreamDescriptor descriptor)
static String	getMongoVersion(MongoDBSourceConfig sourceConfig)
static boolean	isCommandSucceed(org.bson.BsonDocument commandResult)
static org.bson.BsonDocument	isMaster(com.mongodb.client.MongoClient mongoClient)
static io.debezium.relational.TableId	parseTableId(String str)	MongoDB allows dots to be presented in collection names, but not in database names (see docs for more details).
static io.debezium.relational.TableId	parseTableId(String str, boolean useCatalogBeforeSchema)
static List<org.bson.BsonDocument>	readChunks(com.mongodb.client.MongoClient mongoClient, org.bson.BsonDocument collectionMetadata)
static org.bson.BsonDocument	readCollectionMetadata(com.mongodb.client.MongoClient mongoClient, io.debezium.relational.TableId collectionId)
static org.bson.BsonDocument	splitVector(com.mongodb.client.MongoClient mongoClient, io.debezium.relational.TableId collectionId, org.bson.BsonDocument keyPattern, int maxChunkSizeMB)
static org.bson.BsonDocument	splitVector(com.mongodb.client.MongoClient mongoClient, io.debezium.relational.TableId collectionId, org.bson.BsonDocument keyPattern, int maxChunkSizeMB, org.bson.BsonDocument min, org.bson.BsonDocument max)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

COMMAND_SUCCEED_FLAG

public static final org.bson.BsonDouble COMMAND_SUCCEED_FLAG

FAILED_TO_PARSE_ERROR

public static final int FAILED_TO_PARSE_ERROR

See Also:

Constant Field Values

UNAUTHORIZED_ERROR

public static final int UNAUTHORIZED_ERROR

See Also:

Constant Field Values

ILLEGAL_OPERATION_ERROR

public static final int ILLEGAL_OPERATION_ERROR

See Also:

Constant Field Values

CLOSE_CHANGE_STREAM

public static final int CLOSE_CHANGE_STREAM

See Also:

Constant Field Values

RETRY_CHANGE_STREAM

public static final int RETRY_CHANGE_STREAM

See Also:

Constant Field Values

INVALIDATED_RESUME_TOKEN_ERROR

public static final int INVALIDATED_RESUME_TOKEN_ERROR

See Also:

Constant Field Values

CHANGE_STREAM_FATAL_ERROR

public static final int CHANGE_STREAM_FATAL_ERROR

See Also:

Constant Field Values

CHANGE_STREAM_HISTORY_LOST

public static final int CHANGE_STREAM_HISTORY_LOST

See Also:

Constant Field Values

BSON_OBJECT_TOO_LARGE

public static final int BSON_OBJECT_TOO_LARGE

See Also:

Constant Field Values

UNKNOWN_FIELD_ERROR

public static final int UNKNOWN_FIELD_ERROR

See Also:

Constant Field Values

Method Detail

parseTableId

public static io.debezium.relational.TableId parseTableId(String str)

MongoDB allows dots to be presented in collection names, but not in database names (see docs for more details).
So, for a canonical TableId with multiple dots in it (like foo.bar.baz, it is guaranteed that the first part (foo) is the database name and the latter part ( bar.baz) is the table name.

parseTableId

public static io.debezium.relational.TableId parseTableId(String str,
                                                          boolean useCatalogBeforeSchema)

getChangeStreamDescriptor

public static ChangeStreamDescriptor getChangeStreamDescriptor(MongoDBSourceConfig sourceConfig,
                                                               List<String> discoveredDatabases,
                                                               List<String> discoveredCollections)

getChangeStreamIterable

public static com.mongodb.client.ChangeStreamIterable<org.bson.Document> getChangeStreamIterable(MongoDBSourceConfig sourceConfig,
                                                                                                 ChangeStreamDescriptor descriptor)

getChangeStreamIterable

public static com.mongodb.client.ChangeStreamIterable<org.bson.Document> getChangeStreamIterable(com.mongodb.client.MongoClient mongoClient,
                                                                                                 ChangeStreamDescriptor descriptor,
                                                                                                 int batchSize,
                                                                                                 boolean updateLookup,
                                                                                                 boolean fullDocPrePostImage)

getChangeStreamIterable

public static com.mongodb.client.ChangeStreamIterable<org.bson.Document> getChangeStreamIterable(com.mongodb.client.MongoClient mongoClient,
                                                                                                 @Nullable
                                                                                                 String database,
                                                                                                 @Nullable
                                                                                                 String collection,
                                                                                                 @Nullable
                                                                                                 Pattern databaseRegex,
                                                                                                 @Nullable
                                                                                                 Pattern namespaceRegex,
                                                                                                 int batchSize,
                                                                                                 boolean updateLookup,
                                                                                                 boolean fullDocPrePostImage)

getLatestResumeToken

@Nullable
public static org.bson.BsonDocument getLatestResumeToken(com.mongodb.client.MongoClient mongoClient,
                                                         ChangeStreamDescriptor descriptor)

isCommandSucceed

public static boolean isCommandSucceed(org.bson.BsonDocument commandResult)

commandErrorMessage

public static String commandErrorMessage(org.bson.BsonDocument commandResult)

collStats

public static org.bson.BsonDocument collStats(com.mongodb.client.MongoClient mongoClient,
                                              io.debezium.relational.TableId collectionId)

splitVector

public static org.bson.BsonDocument splitVector(com.mongodb.client.MongoClient mongoClient,
                                                io.debezium.relational.TableId collectionId,
                                                org.bson.BsonDocument keyPattern,
                                                int maxChunkSizeMB)

splitVector

public static org.bson.BsonDocument splitVector(com.mongodb.client.MongoClient mongoClient,
                                                io.debezium.relational.TableId collectionId,
                                                org.bson.BsonDocument keyPattern,
                                                int maxChunkSizeMB,
                                                @Nullable
                                                org.bson.BsonDocument min,
                                                @Nullable
                                                org.bson.BsonDocument max)

getCurrentClusterTime

public static org.bson.BsonTimestamp getCurrentClusterTime(com.mongodb.client.MongoClient mongoClient)

isMaster

public static org.bson.BsonDocument isMaster(com.mongodb.client.MongoClient mongoClient)

readChunks

public static List<org.bson.BsonDocument> readChunks(com.mongodb.client.MongoClient mongoClient,
                                                     org.bson.BsonDocument collectionMetadata)

readCollectionMetadata

@Nullable
public static org.bson.BsonDocument readCollectionMetadata(com.mongodb.client.MongoClient mongoClient,
                                                           io.debezium.relational.TableId collectionId)

collectionFor

public static <T> com.mongodb.client.MongoCollection<T> collectionFor(com.mongodb.client.MongoClient mongoClient,
                                                                      io.debezium.relational.TableId collectionId,
                                                                      Class<T> documentClass)

getMongoVersion

public static String getMongoVersion(MongoDBSourceConfig sourceConfig)

clientFor

public static com.mongodb.client.MongoClient clientFor(MongoDBSourceConfig sourceConfig)

buildConnectionString

public static String buildConnectionString(@Nullable
                                           String username,
                                           @Nullable
                                           String password,
                                           String scheme,
                                           String hosts,
                                           @Nullable
                                           String connectionOptions)

checkIfChangeStreamCursorExpires

public static boolean checkIfChangeStreamCursorExpires(com.mongodb.MongoCommandException e)

checkIfResumeTokenExpires

public static boolean checkIfResumeTokenExpires(com.mongodb.MongoCommandException e)