Filepulse S3源连接器报错Provider 's3' not installed，求解决方案

错误原因分析 #

报错FileSystemNotFoundException: Provider "s3" not installed的核心原因有两个：一是缺少S3文件系统NIO提供者依赖，二是连接器配置存在关键错误。

一、配置错误点 #

• 文件读取器选错：当前配置tasks.reader.class用的是LocalRowFileInputReader，这是读取本地文件的实现，S3场景必须替换为S3专用读取器。
• 参数拼写错误：filters.ParseCSVLine.seperator应为separator（少了字母a），会导致CSV解析逻辑失效。

二、需添加的额外依赖库 #

需要将以下JAR包放入Kafka Connect的类路径（如plugin.path指定的目录）：

• aws-java-sdk-s3：AWS S3 SDK核心依赖，版本需与FilePulse兼容
• aws-java-sdk-core：AWS SDK基础核心包
• s3fs-nio：S3的NIO文件系统实现，是解决s3 provider not installed的关键

修正后的完整配置 #

{
  "name": "FilePulseConnector_3",
  "config": {
    "connector.class": "io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector",
    "filters": "ParseCSVLine, Drop",
    "filters.Drop.if": "{{ equals($value.artist, 'U2') }}",
    "filters.Drop.invert": "true",
    "filters.Drop.type": "io.streamthoughts.kafka.connect.filepulse.filter.DropFilter",
    "filters.ParseCSVLine.extract.column.name": "headers",
    "filters.ParseCSVLine.trim.column": "true",
    "filters.ParseCSVLine.separator": ";",
    "filters.ParseCSVLine.type": "io.streamthoughts.kafka.connect.filepulse.filter.DelimitedRowFilter",
    "fs.cleanup.policy.class": "io.streamthoughts.kafka.connect.filepulse.fs.clean.LogCleanupPolicy",
    "fs.cleanup.policy.triggered.on":"COMMITTED",
    "fs.listing.class": "io.streamthoughts.kafka.connect.filepulse.fs.AmazonS3FileSystemListing",
    "fs.listing.filters":"io.streamthoughts.kafka.connect.filepulse.fs.filter.RegexFileListFilter",
    "fs.listing.interval.ms": "10000",
    "file.filter.regex.pattern":".*\\.csv$",
    "offset.policy.class":"io.streamthoughts.kafka.connect.filepulse.offset.DefaultSourceOffsetPolicy",
    "offset.attributes.string": "name",
    "skip.headers": "1",
    "topic": "connect-file-pulse-quickstart-csv",
    "tasks.reader.class": "io.streamthoughts.kafka.connect.filepulse.fs.reader.AmazonS3RowFileInputReader",
    "tasks.file.status.storage.class": "io.streamthoughts.kafka.connect.filepulse.state.KafkaFileObjectStateBackingStore",
    "tasks.file.status.storage.bootstrap.servers": "172.27.157.66:9092",
    "tasks.file.status.storage.topic": "connect-file-pulse-status",
    "tasks.file.status.storage.topic.partitions": 10,
    "tasks.file.status.storage.topic.replication.factor": 1,
    "tasks.max": 1,
    "aws.access.key.id":"<<>>",
    "aws.secret.access.key":"<<>>",
    "aws.s3.bucket.name":"mytestbucketamtrak",
    "aws.s3.region":"us-east-1"
  }
}

补充操作说明 #

1. 依赖包需与FilePulse版本匹配，建议从官方发布包获取对应S3扩展依赖，避免版本冲突。
2. 将依赖JAR放入Kafka Connect的plugin.path下的FilePulse子目录，重启Connect服务生效。

内容的提问来源于stack exchange，提问作者Don Woodward