sanjuthomas / kafka-connect-marklogic Goto Github PK

Apache 2 is common for MarkLogic projects. Example

A CONTRIBUTING.md lets potential contributors know what you expect from them and how they can help. Example

Upgrade writer to give XML document support also.

kafka-connect api
MarkLogic client api
jackson and so on.

Use ML 9 and DMSDK to implement batch write to for sink connect.

EditorConfig is a plugin available for many common editors that will ensure a variety of settings match those used in your project. For instance, this can help make sure that indentation is the way you want it. Example

Implement source connector for MarkLogic

1. enable retry to connect to MarkLogic
2. enable retry/backoff/delay

enable ssl/https connection creation

property - ml.max.retries
fail the task after trying n times - throw a ConnectionException so that the task is shutdown.

On starting the app with Kafka, it is throwing the below errors:

[2018-05-30 16:13:46,276] INFO [Consumer clientId=consumer-8, groupId=connect-marklogic-sink] Revoking previously assigned partitions [] (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:411)
[2018-05-30 16:13:46,276] INFO [Consumer clientId=consumer-8, groupId=connect-marklogic-sink] (Re-)joining group (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:442)
[2018-05-30 16:13:46,276] INFO [Consumer clientId=consumer-9, groupId=connect-marklogic-sink] Revoking previously assigned partitions [] (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:411)
[2018-05-30 16:13:46,279] INFO [Consumer clientId=consumer-9, groupId=connect-marklogic-sink] (Re-)joining group (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:442)
[2018-05-30 16:13:46,289] INFO [Consumer clientId=consumer-8, groupId=connect-marklogic-sink] Successfully joined group with generation 23 (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:409)
[2018-05-30 16:13:46,290] INFO [Consumer clientId=consumer-9, groupId=connect-marklogic-sink] Successfully joined group with generation 23 (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:409)
[2018-05-30 16:13:46,290] INFO [Consumer clientId=consumer-9, groupId=connect-marklogic-sink] Setting newly assigned partitions [] (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:256)
[2018-05-30 16:13:46,290] INFO [Consumer clientId=consumer-8, groupId=connect-marklogic-sink] Setting newly assigned partitions [quote-request-0] (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:256)
[2018-05-30 16:13:46,298] INFO [Consumer clientId=consumer-8, groupId=connect-marklogic-sink] Resetting offset for partition quote-request-0 to offset 0. (org.apache.kafka.clients.consumer.internals.Fetcher:561)
[2018-05-30 16:13:46,303] INFO Received 1 records. kafka coordinates from record: Topic - quote-request, Partition - 0, Offset - 0 (kafka.connect.marklogic.sink.MarkLogicSinkTask:43)
[2018-05-30 16:13:46,310] INFO (withForestConfig) Using [localhost] hosts with forests for "Documents" (com.marklogic.client.datamovement.impl.WriteBatcherImpl:745)
[2018-05-30 16:13:46,310] INFO Adding DatabaseClient on port 8000 for host "localhost" to the rotation (com.marklogic.client.datamovement.impl.WriteBatcherImpl:759)
[2018-05-30 16:13:46,311] INFO threadCount=8 (com.marklogic.client.datamovement.impl.WriteBatcherImpl:259)
[2018-05-30 16:13:46,311] INFO batchSize=100 (com.marklogic.client.datamovement.impl.WriteBatcherImpl:260)
[2018-05-30 16:13:46,311] ERROR WorkerSinkTask{id=marklogic-sink-7} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted. (org.apache.kafka.connect.runtime.WorkerSinkTask:544)
java.lang.NullPointerException
at java.util.HashMap.putMapEntries(HashMap.java:501)
at java.util.LinkedHashMap.(LinkedHashMap.java:384)
at kafka.connect.marklogic.MarkLogicBufferedWriter.lambda$flush$2(MarkLogicBufferedWriter.java:74)
at java.util.ArrayList.forEach(ArrayList.java:1257)
at kafka.connect.marklogic.MarkLogicBufferedWriter.flush(MarkLogicBufferedWriter.java:73)
at kafka.connect.marklogic.MarkLogicBufferedWriter.write(MarkLogicBufferedWriter.java:59)
at kafka.connect.marklogic.sink.MarkLogicSinkTask.put(MarkLogicSinkTask.java:47)
at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:524)
at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:302)
at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:205)
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:173)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)
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)
[2018-05-30 16:13:46,311] INFO Flush - Topic quote-request, Partition 0, Offset 0, Metadata (kafka.connect.marklogic.sink.MarkLogicSinkTask:100)
[2018-05-30 16:13:46,312] ERROR WorkerSinkTask{id=marklogic-sink-7} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:172)
org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:546)
at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:302)
at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:205)
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:173)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)
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.NullPointerException
at java.util.HashMap.putMapEntries(HashMap.java:501)
at java.util.LinkedHashMap.(LinkedHashMap.java:384)
at kafka.connect.marklogic.MarkLogicBufferedWriter.lambda$flush$2(MarkLogicBufferedWriter.java:74)
at java.util.ArrayList.forEach(ArrayList.java:1257)
at kafka.connect.marklogic.MarkLogicBufferedWriter.flush(MarkLogicBufferedWriter.java:73)
at kafka.connect.marklogic.MarkLogicBufferedWriter.write(MarkLogicBufferedWriter.java:59)
at kafka.connect.marklogic.sink.MarkLogicSinkTask.put(MarkLogicSinkTask.java:47)
at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:524)

Add XML message serialization/deserialization support.

In the ml8 version of the connector there is a ml.connection.url property that let's you specifically define the endpoint you'd like to write to.

However, the ml9 version instead has ml.connection.host and ml.connection.port properties, which by default uses the /v1/documents endpoint. The documentation says:
"By default the /v1/documents endpoint at port 8000 is used. You may change that in the marklogic-sink.properties file. You may use your own REST/Service extension instead of the out of the box document API to do any transformation on the way in."

But unless I'm missing something, I can't see a way of defining a REST extension to use instead of the default?