Null Pointer Exception when starting connector

[andy-agap]

Hi, the connector works great on another machine, but when I ported over to a new machine with the same connector configuration, I run into a NullPointerException with the following trace:

[2020-07-07 17:37:32,676] INFO Releasing access on FileStateBackingStore instance for                                                                                                                                                        group fp-test (remaining = null) (io.streamthoughts.kafka.connect.filepulse.state.Stat                                                                                                                                                       eBackingStoreRegistry:82)
[2020-07-07 17:37:32,676] INFO Stopping instance registered instance FileStateBackingS                                                                                                                                                       tore for group fp-test (io.streamthoughts.kafka.connect.filepulse.state.StateBackingSt                                                                                                                                                       oreRegistry:84)
[2020-07-07 17:37:32,676] INFO Closing FileStateBackingStore (io.streamthoughts.kafka.                                                                                                                                                       connect.filepulse.storage.KafkaStateBackingStore:112)
[2020-07-07 17:37:32,676] INFO Stopping KafkaBasedLog for topic connect-file-pulse-sta                                                                                                                                                       tus (io.streamthoughts.kafka.connect.filepulse.storage.KafkaBasedLog:145)
[2020-07-07 17:37:32,677] ERROR WorkerConnector{id=fp-test} Error while starting conne                                                                                                                                                       ctor (org.apache.kafka.connect.runtime.WorkerConnector:119)
java.lang.NullPointerException
at io.streamthoughts.kafka.connect.filepulse.storage.KafkaBasedLog.stop(KafkaBasedLog.java:153)
at io.streamthoughts.kafka.connect.filepulse.storage.KafkaStateBackingStore.stop(KafkaStateBackingStore.java:114)
at io.streamthoughts.kafka.connect.filepulse.state.StateBackingStoreRegistry.release(StateBackingStoreRegistry.java:85)
at io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector.start(FilePulseSourceConnector.java:128)
at org.apache.kafka.connect.runtime.WorkerConnector.doStart(WorkerConnector.java:111)
at org.apache.kafka.connect.runtime.WorkerConnector.start(WorkerConnector.java:136)
at org.apache.kafka.connect.runtime.WorkerConnector.transitionTo(WorkerConnector.java:196)
at org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:252)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:1079)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1300(DistributedHerder.java:117)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:1095)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:1091)
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)

And here is my existing connector config:

{
  "config": {
    "connector.class": "io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector",
    "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy",
    "fs.scanner.class": "io.streamthoughts.kafka.connect.filepulse.scanner.local.LocalFSDirectoryWalker",
    "fs.scan.directory.path": "/path/to/files/",
    "fs.scan.filters": "io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter,io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.LastModifiedFileListFilter",
    "fs.scan.interval.ms": "1000",
    "fs.recursive.scan.enable": "false",
    "file.filter.regex.pattern": "status-[0-9]*.json$",
    "file.filter.minimum.age.ms": "180000",
    "internal.kafka.reporter.bootstrap.servers": "localhost:9092",
    "internal.kafka.reporter.topic": "topic_1",
    "offset.strategy": "name+hash",
    "task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.RowFileInputReader",
    "topic": "topic_2",
    "tasks.max": 1,

	"transforms": "ExtractField",
	"transforms.ExtractField.type":"org.apache.kafka.connect.transforms.ExtractField$Value",
	"transforms.ExtractField.field":"message",

	"key.converter": "org.apache.kafka.connect.json.JsonConverter",
	"value.converter": "org.apache.kafka.connect.storage.StringConverter",
	"key.converter.schemas.enable": "false",
	"value.converter.schemas.enable": "false"

  },
  "name": "filepulse-test"
}

Any advice on what might be causing the issue? Much appreciated.

[fhussonnois]

Hi @andy-agap, sorry for the late reply. To be honest I don't know what may cause this NPE for now. I wil do some investigation to try to reproduce that exception.

[in-tandem]

i am getting the exact same exception, please help

I am gettng it when i run this command
curl -sX POST http://localhost:8083/connectors -d C:\software\kafka_connector_config\connect_file_pulse_risk_etl_topic.json --header "Content-Type: application/json"

[2020-08-03 21:04:27,198] ERROR WorkerConnector{id=connect_file_pulse_risk_etl_topic_4} Error while starting connector (org.apache.kafka.connect.runtime.WorkerConnector)
java.lang.NullPointerException
at io.streamthoughts.kafka.connect.filepulse.storage.KafkaBasedLog.flush(KafkaBasedLog.java:200)
at io.streamthoughts.kafka.connect.filepulse.storage.KafkaStateBackingStore.stop(KafkaStateBackingStore.java:113)
at io.streamthoughts.kafka.connect.filepulse.state.StateBackingStoreRegistry.release(StateBackingStoreRegistry.java:85)
at io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector.start(FilePulseSourceConnector.java:128)
at org.apache.kafka.connect.runtime.Work

trying to run filepulse with below config

{
"config": {
"connector.class": "io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector",
"filters": "GroupMultilineException, ParseLog4jLog",
"filters.GroupMultilineException.negate": "false",
"filters.GroupMultilineException.pattern": "^[\t]",
"filters.GroupMultilineException.type": "io.streamthoughts.kafka.connect.filepulse.filter.MultiRowFilter",
"filters.ParseLog4jLog.match": "%{TIMESTAMP_ISO8601:logdate} %{LOGLEVEL:loglevel} %{GREEDYDATA:message}",
"filters.ParseLog4jLog.overwrite": "message",
"filters.ParseLog4jLog.source": "message",
"filters.ParseLog4jLog.type": "io.streamthoughts.kafka.connect.filepulse.filter.GrokFilter",
"filters.ParseLog4jLog.ignoreFailure": "true",
"fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy",
"fs.scanner.class": "io.streamthoughts.kafka.connect.filepulse.scanner.local.LocalFSDirectoryWalker",
"fs.scan.directory.path": "C:\software\mocked_data\",
"fs.scan.interval.ms": "10000",
"internal.kafka.reporter.bootstrap.servers": "***9092",
"internal.kafka.reporter.id": "connect-file-pulse-log4j-quickstart",
"internal.kafka.reporter.topic": "connect-file-pulse-status",
"offset.strategy": "name",
"read.max.wait.ms": "5000",
"task.reader.class": "io.streamthoughts.kafka.connect.filepulse.reader.RowFileInputReader",
"topic": "risk_etl_topic",
"tasks.max": 1
},
"name": "connect_file_pulse_risk_etl_topic_5"
}

[fhussonnois]

Hi @in-tandem, could you please tell me if you have any log message (with INFO level) starting with Uncaught error from file system monitoring thread ?

[andy-agap]

Hi @in-tandem, I found that creating the topics beforehand helped to solve my NPE problem, Hope it helps.

[fhussonnois]

Hi @in-tandem @andy-agap. I can confirm that an NPE is thrown when the internal.kafka.reporter.topic (default: connect-file-pulse-status) topic is missing. Hence, if auto.create.topics is disable on cluster, the topic must be created before running connector.

I'm will push a commit that will fix the NPE and throw a more meaningful exception

[in-tandem]

thank you so much it worked. to be honest just creating connect-file-pulse-status topic did not work.. i had to additionally setup auto.create.topics.enable=true . are there any other status topics it creates internally.

PS.. @fhussonnois , it is so great to get an answer from you directly :) the only reason i was trying this was because of your great blog. https://medium.com/streamthoughts/kafka-connect-filepulse-one-connector-to-ingest-them-all-faed018a725c. awesome writing

[kbegrow]

@fhussonnois I'm running into this issue since I use 2.0 in some cases. its the same NPE and it is connected with the connect-file-pulse-status & KafkaFileObjectStateBackingStore.
But I have 3-node cluster of connect as well as brokers. The topic gets auto created with 30 partitions and replication factor of 3. (its defined as default from broker side).
This topic is used by 6 connector instances. If I place a new csv file in one of those directories one of the connectors is listing on, this issue happens sporadically.

So far I was not able to find out if its be connected with one of the notes, or the connectors. it seems to be completely random.

Does anyone has any idea?

[fhussonnois]

Hi @kbegrow, could you please create a new issue with the exception stacktrace ? I will investigate to find any NPE cause.