title: How to stream data from Kafka to MongoDB by Kafka Connector
body: <p>I want to stream data from Kafka to MongoDB by using Kafka Connector.
  I found this one <a href="https://github.com/hpgrahsl/kafka-connect-mongodb" rel="nofollow noreferrer">https://github.com/hpgrahsl/kafka-connect-mongodb</a>. But there is no step to do.</p>
  <p>After googling, it seems to lead to Confluent Platform what I don't want to use.</p>
  <p>Could anyone share me document/guideline how to use <a href="https://github.com/hpgrahsl/kafka-connect-mongodb" rel="nofollow noreferrer">kafka-connect-mongodb</a> without using Confluent Platform or another Kafka Connector to stream data from Kafka to MongoDB?</p>
  <p>Thank you in advance.</p>
  <hr />
  <p>What I tried</p>
  <p>Step1: I download <code>mongo-kafka-connect-0.1-all.jar</code> from <a href="https://search.maven.org/search?q=g:org.mongodb.kafka%20AND%20a:mongo-kafka-connect" rel="nofollow noreferrer">maven central</a></p>
  <p>Step2: copy jar file to a new folder <code>plugins</code> inside <code>kafka</code> (I use Kafka on Windows, so the directory is <code>D:\git\1.libraries\kafka_2.12-2.2.0\plugins</code>)</p>
  <p>Step3: Edit file <code>connect-standalone.properties</code> by adding a new line
  <code>plugin.path=/git/1.libraries/kafka_2.12-2.2.0/plugins</code></p>
  <p>Step4: I add new config file for mongoDB sink <code>MongoSinkConnector.properties</code></p>
  <pre><code>name=mongo-sink
  topics=test
  connector.class=com.mongodb.kafka.connect.MongoSinkConnector
  tasks.max=1
  key.ignore=true
  
  # Specific global MongoDB Sink Connector configuration
  connection.uri=mongodb://localhost:27017,mongo1:27017,mongo2:27017,mongo3:27017
  database=test_kafka
  collection=transaction
  max.num.retries=3
  retries.defer.timeout=5000
  type.name=kafka-connect
  </code></pre>
  <p>Step5: run command <code>bin\windows\connect-standalone.bat config\connect-standalone.properties config\MongoSinkConnector.properties</code></p>
  <p>But, I get the error</p>
  <pre><code>[2019-07-09 10:19:09,466] WARN The configuration 'offset.flush.interval.ms' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig)
  [2019-07-09 10:19:09,467] WARN The configuration 'key.converter.schemas.enable' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig)
  [2019-07-09 10:19:09,467] WARN The configuration 'offset.storage.file.filename' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig)
  [2019-07-09 10:19:09,468] WARN The configuration 'value.converter.schemas.enable' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig)
  [2019-07-09 10:19:09,469] WARN The configuration 'plugin.path' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig)
  [2019-07-09 10:19:09,469] WARN The configuration 'value.converter' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig)
  [2019-07-09 10:19:09,470] WARN The configuration 'key.converter' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig)
  Jul 09, 2019 10:19:10 AM org.glassfish.jersey.internal.inject.Providers checkProviderRuntime
  WARNING: A provider org.apache.kafka.connect.runtime.rest.resources.ConnectorPluginsResource registered in SERVER runtime does not implement any provider interfaces applicable in the SERVER runtime. Due to constraint configuration problems the provider org.apache.kafka.connect.runtime.rest.resources.ConnectorPluginsResource will be ignored.
  Jul 09, 2019 10:19:10 AM org.glassfish.jersey.internal.inject.Providers checkProviderRuntime
  WARNING: A provider org.apache.kafka.connect.runtime.rest.resources.RootResource registered in SERVER runtime does not implement any provider interfaces applicable in the SERVER runtime. Due to constraint configuration problems the provider org.apache.kafka.connect.runtime.rest.resources.RootResource will be ignored.
  Jul 09, 2019 10:19:10 AM org.glassfish.jersey.internal.inject.Providers checkProviderRuntime
  WARNING: A provider org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource registered in SERVER runtime does not implement any provider interfaces applicable in the SERVER runtime. Due to constraint configuration problems the provider org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource will be ignored.
  Jul 09, 2019 10:19:11 AM org.glassfish.jersey.internal.Errors logErrors
  WARNING: The following warnings have been detected: WARNING: The (sub)resource method listConnectors in org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource contains empty path annotation.
  WARNING: The (sub)resource method createConnector in org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource contains empty path annotation.
  WARNING: The (sub)resource method listConnectorPlugins in org.apache.kafka.connect.runtime.rest.resources.ConnectorPluginsResource contains empty path annotation.
  WARNING: The (sub)resource method serverInfo in org.apache.kafka.connect.runtime.rest.resources.RootResource contains empty path annotation.
  
  [2019-07-09 10:19:12,302] ERROR WorkerSinkTask{id=mongo-sink-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask)
  org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
          at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:178)
          at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:487)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:464)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:320)
          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: org.apache.kafka.connect.errors.DataException: Converting byte[] to Kafka Connect data failed due to serialization error:
          at org.apache.kafka.connect.json.JsonConverter.toConnectData(JsonConverter.java:344)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$1(WorkerSinkTask.java:487)
          at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128)
          at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162)
          ... 13 more
  Caused by: org.apache.kafka.common.errors.SerializationException: com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'this': was expecting 'null', 'true', 'false' or NaN
   at [Source: (byte[])&quot;this is a message&quot;; line: 1, column: 6]
  Caused by: com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'this': was expecting 'null', 'true', 'false' or NaN
   at [Source: (byte[])&quot;this is a message&quot;; line: 1, column: 6]
          at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:1804)
          at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:703)
          at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._reportInvalidToken(UTF8StreamJsonParser.java:3532)
          at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._reportInvalidToken(UTF8StreamJsonParser.java:3508)
          at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._matchToken2(UTF8StreamJsonParser.java:2843)
          at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._matchTrue(UTF8StreamJsonParser.java:2777)
          at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._nextTokenNotInObject(UTF8StreamJsonParser.java:807)
          at com.fasterxml.jackson.core.json.UTF8StreamJsonParser.nextToken(UTF8StreamJsonParser.java:729)
          at com.fasterxml.jackson.databind.ObjectMapper._readTreeAndClose(ObjectMapper.java:4042)
          at com.fasterxml.jackson.databind.ObjectMapper.readTree(ObjectMapper.java:2571)
          at org.apache.kafka.connect.json.JsonDeserializer.deserialize(JsonDeserializer.java:50)
          at org.apache.kafka.connect.json.JsonConverter.toConnectData(JsonConverter.java:342)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$1(WorkerSinkTask.java:487)
          at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128)
          at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162)
          at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:487)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:464)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:320)
          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)
  [2019-07-09 10:19:12,305] ERROR WorkerSinkTask{id=mongo-sink-0} Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask)
  </code></pre>
  <p>What configuration did I set wrong or I miss anything?</p>
  <hr />
  <p>I fixed it. Now, I can stream data from Kafka to MongoDB succesfully</p>
  <p>My fix is:</p>
  <ol>
  <li>move my kafka to <code>C:\kafka_2.12-2.2.0</code></li>
  <li>update plugin_path corresponding to new path</li>
  <li>update config file <code>connect-standalone.properties</code></li>
  </ol>