我正在研究從postgreSQL到HDFS的數據流 . 我在HDP 2.6沙箱上設置了融合環境 . 我對postgreSQL的jdbc源配置是

name=jdbc_1

connector.class=io.confluent.connect.jdbc.JdbcSourceConnector

tasks.max=1

connection.url=jdbc:postgresql://host:port/db?currentSchema=schema&user=user&password=password

mode=timestamp

timestamp.column.name=col1

validate.non.null=false

topic.prefix=psql-

連接的所有其他屬性也很好，我正在運行它

./bin/connect-standalone ./etc/kafka/connect-standalone.properties ./etc/kafka-connect-jdbc/source.properties

它工作正常，并根據數據庫中的表數創建主題

psql-table1

psql-table2

現在我想在所有主題上運行HDFS接收器，為postgreSQL數據庫中的每個表創建單獨的目錄 . 但是當我用命令運行HDFS接收器時

./bin/connect-standalone ./etc/kafka/connect-standalone.properties ./etc/kafka-connect-hdfs/hdfs-postGres.properties

通過運行源我得到錯誤

ERROR Stopping after connector error (org.apache.kafka.connect.cli.ConnectStandalone:113)

org.apache.kafka.connect.errors.ConnectException: Unable to start REST server

at org.apache.kafka.connect.runtime.rest.RestServer.start(RestServer.java:214)

at org.apache.kafka.connect.runtime.Connect.start(Connect.java:53)

at org.apache.kafka.connect.cli.ConnectStandalone.main(ConnectStandalone.java:95)

Caused by: java.net.BindException: Address already in use

at sun.nio.ch.Net.bind0(Native Method)

at sun.nio.ch.Net.bind(Net.java:433)

at sun.nio.ch.Net.bind(Net.java:425)

at sun.nio.ch.ServerSocketChannelImpl.bind(ServerSocketChannelImpl.java:223)

at sun.nio.ch.ServerSocketAdaptor.bind(ServerSocketAdaptor.java:74)

at org.eclipse.jetty.server.ServerConnector.openAcceptChannel(ServerConnector.java:331)

at org.eclipse.jetty.server.ServerConnector.open(ServerConnector.java:299)

at org.eclipse.jetty.server.AbstractNetworkConnector.doStart(AbstractNetworkConnector.java:80)

at org.eclipse.jetty.server.ServerConnector.doStart(ServerConnector.java:235)

at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:68)

at org.eclipse.jetty.server.Server.doStart(Server.java:398)

at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:68)

at org.apache.kafka.connect.runtime.rest.RestServer.start(RestServer.java:212)

... 2 more

如果我停止源連接并啟動接收器它工作正常 . 任何人都可以幫我，我可以設置多個水槽連接器 .

總結

以上是生活随笔為你收集整理的java 连接kafka_设置多个kafka连接接收器的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。