search

hpgrahsl / kafka-connect-mongodb

**Unofficial / Community** Kafka Connect MongoDB Sink Connector -> integrated 2019 into the official MongoDB Kafka Connector here: https://www.mongodb.com/kafka-connector
Apache License 2.0
153 stars 60 forks source link

Cannot insert a field with '.' ('dot') in field name #78

Closed piyush2206 closed 5 years ago

piyush2206 commented 5 years ago

Hello, Not able to insert records whose field contains '.' (dot) in it. java.lang.IllegalArgumentException: Invalid BSON field name kubernetes.io/config.source

Since MongoDB 3.6, support for key names with dot and dollar sign is allowed. https://docs.mongodb.com/manual/reference/limits/#Restrictions-on-Field-Names

Also tried to work around this by using "post processor - mongodb.field.renamer.regexp" but the post processor (as the name says) does this processing after converting the record to BSON so it breaks before the post-processor itself.

Connector config -

{
  "connector.class": "at.grahsl.kafka.connect.mongodb.MongoDbSinkConnector",
  "topics": "topic-1",
  "mongodb.connection.uri": "mongodb://localhost:27017/test",
  "key.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "key.converter.schemas.enable": "false",
  "value.converter.schemas.enable": "false",
  "mongodb.collection": "mdb-sink-test"
}

Full stack trace - 


org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
    at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:560)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:321)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:224)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:192)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.IllegalArgumentException: Invalid BSON field name scheduler.alpha.kubernetes.io/critical-pod
    at org.bson.AbstractBsonWriter.writeName(AbstractBsonWriter.java:532)
    at org.bson.codecs.BsonDocumentCodec.encode(BsonDocumentCodec.java:114)
    at org.bson.codecs.BsonDocumentCodec.encode(BsonDocumentCodec.java:41)
    at org.bson.codecs.configuration.LazyCodec.encode(LazyCodec.java:37)
    at org.bson.codecs.EncoderContext.encodeWithChildContext(EncoderContext.java:91)
    at org.bson.codecs.BsonDocumentCodec.writeValue(BsonDocumentCodec.java:136)
    at org.bson.codecs.BsonDocumentCodec.encode(BsonDocumentCodec.java:115)
    at org.bson.codecs.BsonDocumentCodec.encode(BsonDocumentCodec.java:41)
    at org.bson.codecs.configuration.LazyCodec.encode(LazyCodec.java:37)
    at org.bson.codecs.EncoderContext.encodeWithChildContext(EncoderContext.java:91)
    at org.bson.codecs.BsonDocumentCodec.writeValue(BsonDocumentCodec.java:136)
    at org.bson.codecs.BsonDocumentCodec.encode(BsonDocumentCodec.java:115)
    at org.bson.codecs.BsonDocumentCodec.encode(BsonDocumentCodec.java:41)
    at org.bson.codecs.BsonDocumentWrapperCodec.encode(BsonDocumentWrapperCodec.java:60)
    at org.bson.codecs.BsonDocumentWrapperCodec.encode(BsonDocumentWrapperCodec.java:29)
    at com.mongodb.operation.BulkWriteBatch$WriteRequestEncoder.encode(BulkWriteBatch.java:398)
    at com.mongodb.operation.BulkWriteBatch$WriteRequestEncoder.encode(BulkWriteBatch.java:377)
    at org.bson.codecs.BsonDocumentWrapperCodec.encode(BsonDocumentWrapperCodec.java:63)
    at org.bson.codecs.BsonDocumentWrapperCodec.encode(BsonDocumentWrapperCodec.java:29)
    at com.mongodb.internal.connection.BsonWriterHelper.writeDocument(BsonWriterHelper.java:75)
    at com.mongodb.internal.connection.BsonWriterHelper.writePayload(BsonWriterHelper.java:59)
    at com.mongodb.internal.connection.CommandMessage.encodeMessageBodyWithMetadata(CommandMessage.java:146)
    at com.mongodb.internal.connection.RequestMessage.encode(RequestMessage.java:138)
    at com.mongodb.internal.connection.CommandMessage.encode(CommandMessage.java:60)
    at com.mongodb.internal.connection.InternalStreamConnection.sendAndReceive(InternalStreamConnection.java:244)
    at com.mongodb.internal.connection.UsageTrackingInternalConnection.sendAndReceive(UsageTrackingInternalConnection.java:99)
    at com.mongodb.internal.connection.DefaultConnectionPool$PooledConnection.sendAndReceive(DefaultConnectionPool.java:444)
    at com.mongodb.internal.connection.CommandProtocolImpl.execute(CommandProtocolImpl.java:72)
    at com.mongodb.internal.connection.DefaultServer$DefaultServerProtocolExecutor.execute(DefaultServer.java:200)
    at com.mongodb.internal.connection.DefaultServerConnection.executeProtocol(DefaultServerConnection.java:269)
    at com.mongodb.internal.connection.DefaultServerConnection.command(DefaultServerConnection.java:131)
    at com.mongodb.operation.MixedBulkWriteOperation.executeCommand(MixedBulkWriteOperation.java:433)
    at com.mongodb.operation.MixedBulkWriteOperation.executeBulkWriteBatch(MixedBulkWriteOperation.java:259)
    at com.mongodb.operation.MixedBulkWriteOperation.access$700(MixedBulkWriteOperation.java:70)
    at com.mongodb.operation.MixedBulkWriteOperation$1.call(MixedBulkWriteOperation.java:203)
    at com.mongodb.operation.MixedBulkWriteOperation$1.call(MixedBulkWriteOperation.java:194)
    at com.mongodb.operation.OperationHelper.withReleasableConnection(OperationHelper.java:424)
    at com.mongodb.operation.MixedBulkWriteOperation.execute(MixedBulkWriteOperation.java:194)
    at com.mongodb.operation.MixedBulkWriteOperation.execute(MixedBulkWriteOperation.java:69)
    at com.mongodb.client.internal.MongoClientDelegate$DelegateOperationExecutor.execute(MongoClientDelegate.java:193)
    at com.mongodb.client.internal.MongoCollectionImpl.executeBulkWrite(MongoCollectionImpl.java:468)
    at com.mongodb.client.internal.MongoCollectionImpl.bulkWrite(MongoCollectionImpl.java:448)
    at at.grahsl.kafka.connect.mongodb.MongoDbSinkTask.processSinkRecords(MongoDbSinkTask.java:148)
    at at.grahsl.kafka.connect.mongodb.MongoDbSinkTask.lambda$null$0(MongoDbSinkTask.java:118)
    at java.util.ArrayList.forEach(ArrayList.java:1257)
    at at.grahsl.kafka.connect.mongodb.MongoDbSinkTask.lambda$put$1(MongoDbSinkTask.java:117)
    at java.util.HashMap.forEach(HashMap.java:1289)
    at at.grahsl.kafka.connect.mongodb.MongoDbSinkTask.put(MongoDbSinkTask.java:112)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:538)
    ... 10 more```
NikhilManapure commented 5 years ago

I am also facing the same issue, I tried using SMT functions but it didn't help out. @hpgrahsl Can you please help me here?

hpgrahsl commented 5 years ago

Thx @piyush2206 and @NikhilManapure for reporting this. In the current version of my sink connector it is not supported to use "." as a field name for several reasons. Very briefly this is why:

1) JSON allows several field name characters which are either only partially supported or not valid to be used in MongoDB documents at all. By that I mean that while it would be possible to convince the db to store field names containing "." anyway, it is IMHO not recommended to do so... at least not at the moment since you cannot query such fields properly in order to read your data back. probably the best source to read about this which is also reflecting my personal opinion why there is no support for this can be found in related tickets at the official project:

2) Kafka Connect and hence my sink connector support also AVRO encoded records besides JSON. and the AVRO spec is much more restrictive with regard to field names - so it didn't make too much sense for me to support things for one encoding which isn't possible within another. read the relevant section in the AVRO spec in case you want to dig deeper here https://avro.apache.org/docs/1.8.1/spec.html#names

3) other than the mere writing of documents with "." characters in field names, there are a lot of configuration properties for my sink connector which are using "." as special character e.g. in order to navigate the nesting of the document structure itself - this would clash and become unusable with field names containing "." at the moment. while this might change in future releases it most likely won't before an officially supported solution has been found for the mongodb server and the drivers respectively.

I hope this clarifies the considerations that went into the current "design" in the first place. Given this the best way to go IMHO is to solve this "upstream" e.g. during a stream processing step or trying to use SMT and replace all "." dots in your field names (recursively) so that your documents are "clean" before they are handed to the sink connector.

Happy to hear your thoughts!