Skip to content

feat: custom schema #30

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 6 commits into
base: master
Choose a base branch
from
Open

feat: custom schema #30

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
5bfffcc
feat: custom schema
miguelmarmol Jul 31, 2018
5cab9a4
remove blank space MongodSourceTask
miguelmarmol Jul 31, 2018
a51c028
make 'custom.schema' property optional
miguelmarmol Aug 1, 2018
1fa5fe2
set messages to debug level
miguelmarmol Aug 6, 2018
0761793
support undefined type
miguelmarmol Aug 6, 2018
1ecb02a
support null values
miguelmarmol Aug 6, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 12 additions & 8 deletions src/main/java/org/apache/kafka/connect/mongodb/MongodbSourceConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,15 +29,19 @@ public class MongodbSourceConfig extends AbstractConfig {
public static final String CONVERTER_CLASS = "converter.class";
private static final String CONVERTER_CLASS_DOC = "Converter class used to transform a mongodb oplog in a kafka message";

public static final String CUSTOM_SCHEMA = "custom.schema";
private static final String CUSTOM_SCHEMA_DOC = "Flag that tells if a custom schema will be obtained from the data";

public static ConfigDef config = new ConfigDef()
.define(URI, Type.STRING, Importance.HIGH, URI_DOC)
.define(HOST, Type.STRING, Importance.HIGH, HOST_DOC)
.define(PORT, Type.INT, Importance.HIGH, PORT_DOC)
.define(BATCH_SIZE, Type.INT, Importance.HIGH, BATCH_SIZE_DOC)
.define(SCHEMA_NAME, Type.STRING, Importance.HIGH, SCHEMA_NAME_DOC)
.define(TOPIC_PREFIX, Type.STRING, Importance.LOW, TOPIC_PREFIX_DOC)
.define(CONVERTER_CLASS, Type.STRING, StringStructConverter.class.getName(), Importance.LOW, CONVERTER_CLASS_DOC)
.define(DATABASES, Type.STRING, Importance.LOW, DATABASES_DOC);
.define(URI, Type.STRING, Importance.HIGH, URI_DOC)
.define(HOST, Type.STRING, Importance.HIGH, HOST_DOC)
.define(PORT, Type.INT, Importance.HIGH, PORT_DOC)
.define(BATCH_SIZE, Type.INT, Importance.HIGH, BATCH_SIZE_DOC)
.define(SCHEMA_NAME, Type.STRING, Importance.HIGH, SCHEMA_NAME_DOC)
.define(TOPIC_PREFIX, Type.STRING, Importance.LOW, TOPIC_PREFIX_DOC)
.define(CUSTOM_SCHEMA, Type.BOOLEAN, false, Importance.LOW, CUSTOM_SCHEMA_DOC)
.define(CONVERTER_CLASS, Type.STRING, StringStructConverter.class.getName(), Importance.LOW, CONVERTER_CLASS_DOC)
.define(DATABASES, Type.STRING, Importance.LOW, DATABASES_DOC);

public MongodbSourceConfig(Map<String, String> props) {
super(config, props);
Expand Down
26 changes: 18 additions & 8 deletions src/main/java/org/apache/kafka/connect/mongodb/MongodbSourceConnector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import static org.apache.kafka.connect.mongodb.MongodbSourceConfig.TOPIC_PREFIX;
import static org.apache.kafka.connect.mongodb.MongodbSourceConfig.CONVERTER_CLASS;
import static org.apache.kafka.connect.mongodb.MongodbSourceConfig.DATABASES;
import static org.apache.kafka.connect.mongodb.MongodbSourceConfig.CUSTOM_SCHEMA;

/**
* MongodbSourceConnector implements the connector interface
Expand All @@ -39,6 +40,7 @@ public class MongodbSourceConnector extends SourceConnector {
private String topicPrefix;
private String converterClass;
private String databases;
private String customSchema;

/**
* Get the version of this connector.
Expand All @@ -64,14 +66,14 @@ public void start(Map<String, String> map) {
if (uri == null || uri.isEmpty()){
host = map.get(HOST);
if (host == null || host.isEmpty()){
throw new ConnectException("Missing " + HOST + "or " + URI + " config");
throw new ConnectException("Missing " + HOST + "or " + URI + " config");
}
port = map.get(PORT);

port = map.get(PORT);
if (port == null || port.isEmpty()){
throw new ConnectException("Missing " + PORT + "or " + URI + " config");
}
}
}
schemaName = map.get(SCHEMA_NAME);
if (schemaName == null || schemaName.isEmpty())
throw new ConnectException("Missing " + SCHEMA_NAME + " config");
Expand All @@ -83,9 +85,15 @@ public void start(Map<String, String> map) {
databases = map.get(DATABASES);

topicPrefix = map.get(TOPIC_PREFIX);

converterClass = map.get(CONVERTER_CLASS);

if(map.containsKey(CUSTOM_SCHEMA))
customSchema = map.get(CUSTOM_SCHEMA);
else
customSchema = null;


LogUtils.dumpConfiguration(map, log);
}

Expand Down Expand Up @@ -115,16 +123,18 @@ public List<Map<String, String>> taskConfigs(int maxTasks) {
Map<String, String> config = new HashMap<>();
config.put(URI, uri);
if(host!=null){
config.put(HOST, host);
config.put(HOST, host);
}
if(port!=null){
config.put(PORT, port);
config.put(PORT, port);
}
config.put(SCHEMA_NAME, schemaName);
config.put(BATCH_SIZE, batchSize);
config.put(TOPIC_PREFIX, topicPrefix);
config.put(CONVERTER_CLASS, converterClass);
config.put(DATABASES, StringUtils.join(dbsGrouped.get(i), ","));
if(customSchema != null)
config.put(CUSTOM_SCHEMA, customSchema);
configs.add(config);
}
return configs;
Expand All @@ -143,4 +153,4 @@ public ConfigDef config () {
}


}
}
146 changes: 99 additions & 47 deletions src/main/java/org/apache/kafka/connect/mongodb/MongodbSourceTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,12 +1,6 @@
package org.apache.kafka.connect.mongodb;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.mongodb.MongoClient;
import org.apache.kafka.common.errors.InterruptException;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
Expand All @@ -16,11 +10,16 @@
import org.apache.kafka.connect.mongodb.converter.StructConverter;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.source.SourceTask;
import org.apache.kafka.connect.utils.ConverterUtils;
import org.bson.BsonDocument;
import org.bson.BsonTimestamp;
import org.bson.BsonValue;
import org.bson.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

/**
* MongodbSourceTask is a Task that reads mutations from a mongodb for storage in Kafka.
*
Expand All @@ -41,6 +40,8 @@ public class MongodbSourceTask extends SourceTask {

private MongodbReader reader;

private Boolean customSchema = false;


Map<Map<String, String>, Map<String, Object>> offsets = new HashMap<>(0);

Expand All @@ -57,14 +58,14 @@ public String version() {
*/
@Override
public void start(Map<String, String> map) {
if(map.containsKey(MongodbSourceConfig.PORT)){
try {
port = Integer.parseInt(map.get(MongodbSourceConfig.PORT));
} catch (Exception e) {
throw new ConnectException(MongodbSourceConfig.PORT + " config should be an Integer");
}
}
if(map.containsKey(MongodbSourceConfig.PORT)){
try {
port = Integer.parseInt(map.get(MongodbSourceConfig.PORT));
} catch (Exception e) {
throw new ConnectException(MongodbSourceConfig.PORT + " config should be an Integer");
}
}

try {
batchSize = Integer.parseInt(map.get(MongodbSourceConfig.BATCH_SIZE));
} catch (Exception e) {
Expand All @@ -75,18 +76,26 @@ public void start(Map<String, String> map) {
topicPrefix = map.get(MongodbSourceConfig.TOPIC_PREFIX);
uri = map.get(MongodbSourceConfig.URI);
host = map.get(MongodbSourceConfig.HOST);

try{
String structConverterClass = map.get(MongodbSourceConfig.CONVERTER_CLASS);
if(structConverterClass == null || structConverterClass.isEmpty()){
structConverterClass = StringStructConverter.class.getName();
structConverterClass = StringStructConverter.class.getName();
}
structConverter = (StructConverter) Class.forName(structConverterClass).newInstance();
}
catch(Exception e){
throw new ConnectException(MongodbSourceConfig.CONVERTER_CLASS + " config should be a class of type StructConverter");
throw new ConnectException(MongodbSourceConfig.CONVERTER_CLASS + " config should be a class of type StructConverter");
}

try{
if(map.containsKey(MongodbSourceConfig.CUSTOM_SCHEMA))
customSchema = Boolean.parseBoolean(map.get(MongodbSourceConfig.CUSTOM_SCHEMA));
}

catch (Exception e){
throw new ConnectException(MongodbSourceConfig.CUSTOM_SCHEMA + " config should be a Boolean");
}

databases = Arrays.asList(map.get(MongodbSourceConfig.DATABASES).split(","));

log.trace("Creating schema");
Expand All @@ -98,24 +107,24 @@ public void start(Map<String, String> map) {
db = db.replaceAll("[\\s.]", "_");
if (schemas.get(db) == null)
schemas.put(db,
SchemaBuilder
.struct()
.name(schemaName.concat("_").concat(db))
.field("timestamp", Schema.OPTIONAL_INT32_SCHEMA)
.field("order", Schema.OPTIONAL_INT32_SCHEMA)
.field("operation", Schema.OPTIONAL_STRING_SCHEMA)
.field("database", Schema.OPTIONAL_STRING_SCHEMA)
.field("object", Schema.OPTIONAL_STRING_SCHEMA)
.build());
SchemaBuilder
.struct()
.name(schemaName.concat("_").concat(db))
.field("timestamp", Schema.OPTIONAL_INT32_SCHEMA)
.field("order", Schema.OPTIONAL_INT32_SCHEMA)
.field("operation", Schema.OPTIONAL_STRING_SCHEMA)
.field("database", Schema.OPTIONAL_STRING_SCHEMA)
.field("object", Schema.OPTIONAL_STRING_SCHEMA)
.build());
}

loadOffsets();

if(uri != null){
reader = new MongodbReader(uri, databases, batchSize, offsets);
reader = new MongodbReader(uri, databases, batchSize, offsets);
}
else{
reader = new MongodbReader(host, port, databases, batchSize, offsets);
reader = new MongodbReader(host, port, databases, batchSize, offsets);
}
reader.run();
}
Expand All @@ -130,12 +139,55 @@ public void start(Map<String, String> map) {
public List<SourceRecord> poll() throws InterruptException {
List<SourceRecord> records = new ArrayList<>();
while (!reader.isEmpty()) {
Document message = reader.pool();
Document message = reader.pool();
Struct messageStruct = getStruct(message);
String topic = getTopic(message);
String db = getDB(message);
String timestamp = getTimestamp(message);
records.add(new SourceRecord(Collections.singletonMap("mongodb", db), Collections.singletonMap(db, timestamp), topic, messageStruct.schema(), messageStruct));
Object objectKey = ((Map<String, Object>) message.get("o")).get("_id");

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you please clean up the extraneous whitespace here and at line 152?

//key is _id
if(customSchema){
//get the key object
BsonDocument bson = message
.toBsonDocument(
BsonDocument.class, MongoClient.getDefaultCodecRegistry());
BsonValue bsonId = bson.get("o").asDocument().get("_id");
Schema keySchema = null;
Object key = null;
if(bsonId.isObjectId()){
keySchema = Schema.STRING_SCHEMA;
key = bsonId.asObjectId().getValue().toString();
}
else if(bsonId.isDocument()){
keySchema = ConverterUtils.createDynamicSchema((Map<String, Object>) objectKey, false);
key = ConverterUtils.createDynamicStruct(keySchema, (Map<String, Object>) objectKey, false);
}
else{
keySchema = Schema.STRING_SCHEMA;
key = String.valueOf(objectKey);
}

records.add(new SourceRecord(
Collections.singletonMap("mongodb", db),
Collections.singletonMap(db, timestamp),
topic,
keySchema,
key,
messageStruct.schema(),
messageStruct));
}

//key null
else {
records.add(new SourceRecord(
Collections.singletonMap("mongodb", db),
Collections.singletonMap(db, timestamp),
topic,
messageStruct.schema(),
messageStruct)
);
}
log.trace(message.toString());
}

Expand All @@ -148,9 +200,9 @@ public List<SourceRecord> poll() throws InterruptException {
*/
@Override
public void stop() {
if(reader != null){
reader.stop();
}
if(reader != null){
reader.stop();
}
}

/**
Expand All @@ -163,10 +215,10 @@ private String getTopic(Document message) {
String database = ((String) message.get("ns")).replaceAll("[\\s.]", "_");
if (topicPrefix != null && !topicPrefix.isEmpty()) {
return new StringBuilder()
.append(topicPrefix)
.append("_")
.append(database)
.toString();
.append(topicPrefix)
.append("_")
.append(database)
.toString();
}
return database;
}
Expand All @@ -190,10 +242,10 @@ private String getDB(Document message) {
private String getTimestamp(Document message) {
BsonTimestamp timestamp = (BsonTimestamp) message.get("ts");
return new StringBuilder()
.append(timestamp.getTime())
.append("_")
.append(timestamp.getInc())
.toString();
.append(timestamp.getTime())
.append("_")
.append(timestamp.getInc())
.toString();
}

/**
Expand All @@ -203,8 +255,8 @@ private String getTimestamp(Document message) {
* @return message formatted as a Struct
*/
private Struct getStruct(Document message) {
final Schema schema = schemas.get(getDB(message).replaceAll("[\\s.]", "_"));
return structConverter.toStruct(message, schema);
final Schema schema = schemas.get(getDB(message).replaceAll("[\\s.]", "_"));
return structConverter.toStruct(message, schema, customSchema);
}

/**
Expand All @@ -218,4 +270,4 @@ private void loadOffsets() {
}
offsets.putAll(context.offsetStorageReader().offsets(partitions));
}
}
}
6 changes: 3 additions & 3 deletions src/main/java/org/apache/kafka/connect/mongodb/converter/JsonStructConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,13 +7,13 @@

/**
* Struct converter who stores mongodb document as Json String.
*
*
* @author André Ignacio
*/
public class JsonStructConverter implements StructConverter {

@Override
public Struct toStruct(Document document, Schema schema) {
public Struct toStruct(Document document, Schema schema, Boolean customSchema) {
final Struct messageStruct = new Struct(schema);
final BsonTimestamp bsonTimestamp = (BsonTimestamp) document.get("ts");
final Integer seconds = bsonTimestamp.getTime();
Expand All @@ -22,7 +22,7 @@ public Struct toStruct(Document document, Schema schema) {
messageStruct.put("order", order);
messageStruct.put("operation", document.get("op"));
messageStruct.put("database", document.get("ns"));

final Document modifiedDocument = (Document) document.get("o");
messageStruct.put("object", modifiedDocument.toJson());

Expand Down
Loading