HDFS connector with cluster in HA mode about kafka-connect-hdfs

[Ishiihara]

@wjptak The connector should work with Hadoop HA. Here is a discussion on the connector works with HA: https://groups.google.com/forum/?utm_medium=email&utm_source=footer#!msg/confluent-platform/VPS79sqe84I/e74l_Gg-AgAJ

[Ishiihara]

Closing this because it is not an issue.

[wjptak]

@Ishiihara that's great news, we'll try this.

May I suggest to put this in the documentation? That would be very useful!

Thanks!

[drdee]

@Ishiihara does that mean that https://github.com/confluentinc/kafka-connect-hdfs/blob/master/src/main/java/io/confluent/connect/hdfs/HdfsSinkConnectorConfig.java#L40-L42 is outdated?

[jecos]

@Ishiihara I tried to use the HA name of the clusters in hdfs.url but i got this error :

INFO Couldn't start HdfsSinkConnector: (io.confluent.connect.hdfs.HdfsSinkTask) org.apache.kafka.connect.errors.ConnectException: java.lang.reflect.InvocationTargetException at io.confluent.connect.hdfs.storage.StorageFactory.createStorage(StorageFactory.java:31) at io.confluent.connect.hdfs.DataWriter.<init>(DataWriter.java:168) at io.confluent.connect.hdfs.HdfsSinkTask.start(HdfsSinkTask.java:64) at org.apache.kafka.connect.runtime.WorkerSinkTask.initializeAndStart(WorkerSinkTask.java:207) at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:139) at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:140) at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:175) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) Caused by: java.lang.reflect.InvocationTargetException at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62) at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) at java.lang.reflect.Constructor.newInstance(Constructor.java:422) at io.confluent.connect.hdfs.storage.StorageFactory.createStorage(StorageFactory.java:29) ... 11 more Caused by: java.lang.IllegalArgumentException: java.net.UnknownHostException: nameservice1 at org.apache.hadoop.security.SecurityUtil.buildTokenService(SecurityUtil.java:374) at org.apache.hadoop.hdfs.NameNodeProxies.createNonHAProxy(NameNodeProxies.java:312) at org.apache.hadoop.hdfs.NameNodeProxies.createProxy(NameNodeProxies.java:178) at org.apache.hadoop.hdfs.DFSClient.<init>(DFSClient.java:665) at org.apache.hadoop.hdfs.DFSClient.<init>(DFSClient.java:601) at org.apache.hadoop.hdfs.DistributedFileSystem.initialize(DistributedFileSystem.java:148) at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2596) at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:91) at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:2630) at org.apache.hadoop.fs.FileSystem$Cache.getUnique(FileSystem.java:2618) at org.apache.hadoop.fs.FileSystem.newInstance(FileSystem.java:417) at io.confluent.connect.hdfs.storage.HdfsStorage.<init>(HdfsStorage.java:39) ... 16 more Caused by: java.net.UnknownHostException: nameservice1 ... 28 more

In the constructor of HdfsStorage, fs is initialize with :

fs = FileSystem.newInstance(URI.create(url), conf);

It shouldn't use systematically this helper but rather this one :

fs = FileSystem.newInstance(conf);

In other words, use what is defined in the config hdfs-site.xml to init hdfs client

[rnpridgeon]

@jecos make sure you have set hadoop.conf.dir to a directory local to the worker which contains both hdfs-site.xml and core-site.xml. hdfs-site.xml provides all the necessary configs to utilize the ConfiguredFailoverProxyProvider while core-site.xml provides the value of defaultFS.

As long as both of these files are properly configured for HA HDFS and the task is able to pick them up you will be able to reference the nameservice id.

[jecos]

@rnpridgeon this is exactly what i done. But still get java.net.UnknownHostException: nameservice1.

This is what i have in the hdfs-site.xml :

  <property>
    <name>dfs.nameservices</name>
    <value>nameservice1</value>
  </property>
  <property>
    <name>dfs.client.failover.proxy.provider.nameservice1</name>
    <value>org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider</value>
  </property>

And I configured the connector with :

     "hadoop.conf.dir": "/etc/hadoop/conf",
    "hdfs.url": "hdfs://nameservice1",