KafkaConnector - FilePulse error

[jkleinkauff]

Hello folks.

I'm trying to use a KafkaConnector (FilePulse) without success. I'm unsure if I may be doing something wrong or it's a bug, probably not.

After applying the below configs, i can see two issues form the connector:

k describe kafkaconnector my-source-connector -n kafka

After a few minutes, the same command returns

kafka.yaml

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    replicas: 1
    logging:
      type: inline
      loggers:
        kafka.root.logger.level: "DEBUG"
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: scram-sha-512
      - name: external
        port: 9094
        type: ingress
        tls: true
        authentication:
          type: scram-sha-512
        configuration:
          class: nginx
          bootstrap:
            host: bootstrap.192.168.15.160.nip.io
          brokers:
          - broker: 0
            host: broker-0.192.168.15.160.nip.io
    authorization:
      type: simple
    storage:
      deleteClaim: true
      size: 10Gi
      type: persistent-claim
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
      default.replication.factor: 1
      log.message.format.version: "2.1"
  zookeeper:
    replicas: 1
    storage:
      type: persistent-claim
      size: 10Gi
      deleteClaim: true
  entityOperator:
    topicOperator: {}
    userOperator: {}

connect.yaml

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  version: 3.2.0
  replicas: 1
  bootstrapServers: my-cluster-kafka-bootstrap:9093
  tls:
    trustedCertificates:
      - secretName: my-cluster-cluster-ca-cert
        certificate: ca.crt
  authentication:
    type: scram-sha-512
    username: my-connect
    passwordSecret:
      secretName: my-connect
      password: password
  config:
    group.id: connect-cluster
    replication.factor: 1
    default.replication.factor: 1
    offsets.topic.replication.factor: 1
    transaction.state.log.replication.factor: 1
    offset-syncs.topic.replication.factor: 1
    config.storage.replication.factor: 1
    offset.storage.replication.factor: 1
    status.storage.replication.factor: 1
    offset.storage.topic: connect-cluster-offsets
    config.storage.topic: connect-cluster-configs
    status.storage.topic: connect-cluster-status
  build:
    output:
      type: docker
      image: <myuser>/my-connect-cluster:latest
      pushSecret: redcred
    plugins:
      - name: debezium-postgres-connector
        artifacts:
          - type: zip
            url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/1.4.2.Final/debezium-connector-postgres-1.4.2.Final-plugin.zip
            sha512sum: ef1620547e6ddf5be010271849b6a87a19f6e6beee93b379c80883815b8f37ec5137095b2c99975d7704cbf957e6a33d76c61109582cad57c7cbbfae43adc86c
      - name: filepulse-connector
        artifacts:
          - type: zip
            url: https://github.com/streamthoughts/kafka-connect-file-pulse/releases/download/v2.6.0/streamthoughts-kafka-connect-file-pulse-2.6.0.zip
  template:
    pod:
      securityContext:
        runAsUser: 0

connector.yaml

---
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: my-source-connector
  labels:
    strimzi.io/cluster: my-connect
spec:
  class: io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector
  tasksMax: 2
  config:
    tasks.reader.class: "io.streamthoughts.kafka.connect.filepulse.fs.reader.LocalRowFileInputReader"
    fs.scan.filters: "io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter"
    fs.cleanup.policy.class: "io.streamthoughts.kafka.connect.filepulse.fs.clean.LogCleanupPolicy"
    fs.listing.class: "io.streamthoughts.kafka.connect.filepulse.fs.LocalFSDirectoryListing"
    fs.listing.directory.path: "/tmp/"
    fs.scan.interval.ms: "10000"
    file.filter.regex.pattern: ".*\\.csv$"
    offset.strategy: "name"
    skip.headers: "1"
    topic: "musics-filepulse-csv-222"
    internal.kafka.reporter.topic: "connect-file-pulse-status"
    tasks.file.status.storage.bootstrap.servers: "my-cluster-kafka-bootstrap:9093"

Tail from the connect pod:

Can someone guide me in the right direction? Thank you!

[scholzj]

Do you have any other working connectors in the cluster? It seems it has issues connecting to Kafka. But its hard to say without the full logs and full custom resources including the statuses.

[scholzj]

Yeah, that would be the issue. Why does the connector have this configuration option?

tasks.file.status.storage.bootstrap.servers: "my-cluster-kafka-bootstrap:9093"

It sounds like the connector opens its own Kafka client with its own Connections instead of using Connect. So you might need to configure it.

[jkleinkauff]

Understood, thank you so much!

Just out of curiosity, it's not common for the Connector to have this option? I mean, from your perspective, there should be another config to make this work?

[scholzj]

The main point of Kafka Connect is that Kafka Connect takes care of the Kafka side and the connector plugin takes care of the other system it is connecting to. So form my experience, most other connectors don't do that. I know only of one other connector which is connecting to Kafka also directly which is Debezium.

[jkleinkauff]

Look who i found here.
So do you still recommend the built-in FileStreamSource to ingest CSV files ?

[scholzj]

Long time ago ... the FileStreamSource is in a way an example with very limited features. But it works, yes. That said, the connector might have its reasons why it does what it does -> Debezium certainly does. So I'm not saying it is necessarily wrong ... just that it is a little bit unusual and can make things a bit complicated.

That said, Strimzi based Kafka Connect currently does not support any volume mounts. So I'm not sure how any reading from file would work with Strimzi.