Kafka对接Minio配置步骤咨询及端口占用报错处理

错误日志显示java.net.BindException: Address already in use，说明8083端口被其他进程占用，解决方法如下：

• 查找占用8083端口的进程：
  • Linux/macOS：执行 lsof -i :8083 或 netstat -tulpn | grep 8083 获取进程ID
  • Windows：执行 netstat -ano | findstr :8083 获取进程ID
• 终止占用进程：
  • Linux/macOS：执行 kill -9 <进程ID>
  • Windows：执行 taskkill /F /PID <进程ID>
• （可选）修改Kafka Connect端口：
  在Connect配置文件（connect-standalone.properties 或 connect-distributed.properties）中，修改 rest.port 参数为未被占用的端口（如 rest.port=8084），然后重启Kafka Connect

Minio兼容S3 API，直接复用Amazon S3 Sink Connector即可，配置步骤如下：

1. 编写Connector配置文件 #

创建配置文件（例如 kafka-minio-sink.properties），填入以下核心配置：

name=kafka-minio-sink
connector.class=io.confluent.connect.s3.S3SinkConnector
tasks.max=1
topics=需要同步的Kafka主题名称

# Minio连接配置
s3.region=us-east-1  # Minio默认区域，可自定义
s3.endpoint=http://Minio服务器地址:端口  # 示例：http://localhost:9000
s3.access.key.id=Minio访问密钥
s3.secret.access.key=Minio秘密密钥
s3.path.style.access=true  # 必须开启，Minio使用路径式存储访问

# 存储格式与滚动策略（按需调整）
format.class=io.confluent.connect.s3.format.json.JsonFormat
flush.size=1000  # 累计1000条消息写入一个文件
rotate.interval.ms=3600000  # 每小时自动滚动生成新文件
storage.class=io.confluent.connect.s3.storage.S3Storage
partitioner.class=io.confluent.connect.storage.partitioner.DefaultPartitioner

2. 启动Connector #

• Standalone模式：执行命令

bin/connect-standalone.sh config/connect-standalone.properties config/kafka-minio-sink.properties

• Distributed模式：将配置转为JSON格式后，通过REST API提交

curl -X POST -H "Content-Type: application/json" --data @kafka-minio-sink.json http://localhost:8083/connectors

3. 验证同步结果 #

• 检查Connector运行状态：curl http://localhost:8083/connectors/kafka-minio-sink/status
• 登录Minio控制台，查看对应存储桶下是否生成了Kafka主题的消息文件

内容的提问来源于stack exchange，提问作者hasna elghyaby