# HDFS 2 Source Connector for Confluent Platform [Deprecated] #### IMPORTANT This connector is deprecated and will reach its end of life (EOL) on the Confluent Platform 7.9 end of support (EOS) date. Confluent recommends migrating to [HDFS 3 Source connector](https://docs.confluent.io/kafka-connectors/hdfs3-source/current/overview.html) before the EOL date. For more information, see [Connector support lifecycle policy](https://docs.confluent.io/platform/current/connect/supported.html#support-policy-for-self-managed-connectors). The Kafka Connect HDFS 2 Source connector provides the capability to read data exported to HDFS2 by the Kafka [HDFS 2 Sink connector for Confluent Platform](https://docs.confluent.io/kafka-connect-hdfs/current/index.html) and publish it back to a Kafka topic. Note the connector only works for HDFS 2 objects created by the [HDFS 2 Sink connector for Confluent Platform](https://docs.confluent.io/kafka-connect-hdfs/current/index.html) and is released separately from the HDFS 3.x connector. If you are targeting an HDFS 3.x distribution, see the [HDFS 3 Source connector for Confluent Platform](https://docs.confluent.io/kafka-connect-hdfs3-source/current/) documentation for more details. Depending on the format and partitioner used to write the data to HDFS 2, this connector can write to the destination topic using the same partitions as the original messages exported to HDFS 2 Sink and maintain the same message order. The connector selects folders based on the partitioner configuration and reads each folder’s HDFS 2 objects in alphabetical order. Each record is read based on the format selected. Configuration is set up to mirror the Kafka [HDFS 2 Sink Connector for Confluent Platform](https://docs.confluent.io/kafka-connect-hdfs/current/index.html) and should be possible to make only minor changes to the original sink configuration. #### IMPORTANT - You can create topics manually in the destination Kafka cluster with the correct number of partitions before running the source connector. If the topics do not exist, Connect relies on [Configuring Auto Topic Creation for Source Connectors](/kafka-connectors/self-managed/userguide.html#connect-source-auto-topic-creation) and the number of partitions are based upon the Kafka broker defaults. If there are more partitions in the destination cluster, the extra partitions are not used. If there are fewer partitions in the destination cluster, the connector task throws an exception and stops the moment it tries to write to a Kafka partition that does not exist. - The connector ignores any HDFS 2 object with a name that does not start with the configured topics directory. This name is `"/topics/"` by default. - The connector ignores any HDFS 2 object that is below the topics directory but has an extension that does not match the configured format. For example, a JSON file is ignored when `format.class` is set for Avro files. - The connector stops and fails if the HDFS 2 object’s name does not match the expected format or is in an unexpected location. - Effective July 6, 2025, only self-managed connector versions that meet or exceed the minimum version listed on the [Supported Connector Versions](https://docs.confluent.io/platform/7.8/connect/supported-connector-version.html#supported-connector-versions-till-cp-7-8) page receive support from Confluent. Older, unsupported connector versions have been removed from Confluent Marketplace and are no longer available for download. Check your configuration to avoid the following issues: * A file with the correct extension and a valid name format (for example, `+++.`) placed in a folder of a different topic will be read normally and written to whatever topic as defined by its filename. * If a field partitioner is incorrectly configured to match the expected folder, it can break the ordering guarantees of the HDFS 2 sink that used a deterministic sink partitioner. ## Features The HDFS 2 Source connector includes the following features: - [Pluggable data format with or without schema](#hdfs2-source-pluggable-data-format-with-or-without-schema) - [At least once delivery](#hdfs2-source-atleast-once-delivery) - [Multiple tasks](#hdfs2-source-multiple-tasks) - [Matching source partitioning](#hdfs2-match-source-partitioning) - [Source partition ordering](#hdfs2-source-partition-ordering) - [Pluggable partitioner](#hdfs2-source-pluggable-partitioner) ### Pluggable data format with or without schema Out of the box, the connector supports reading data from HDFS 2 in Avro and JSON format. Besides records with schema, the connector supports importing plain JSON records without schema in text files, one record per line. In general, the connector may accept any format that provides an implementation of the Format interface. ### At least once delivery In the event of a task failure the connector guarantees no messages are lost. The connector manages offsets so that it can start from the last committed offsets in case of failures and task restarts. ### Multiple tasks The HDFS2 Source connector supports running one or more tasks. You can specify the number of tasks in the `tasks.max` configuration parameter. This can lead to huge performance gains when multiple files need to be parsed. ### Matching source partitioning Messages will be put back on to the same Kafka partition for that topic when it was written. ### Source partition ordering The connector will read records back in time order in each topic-source partition if the `DefaultPartitioner` or a `TimeBasedPartitioner` is used. If a `FieldPartitioner` is used it isn’t possible to guarantee the order of these messages. ### Pluggable partitioner The connector comes out of the box with partitioners that support default partitioning based on Kafka partitions, field partitioning, and time-based partitioning in days or hours. You may implement your own partitioners by extending the Partitioner class. Additionally, you can customize time based partitioning by extending the TimeBasedPartitioner class. ## Limitations For the `TimeBasedPartitioner`, the capacity to scale the connector across various time ranges is limited. Currently, the connector does not support processing data that spans several years. ## License You can use this connector for a 30-day trial period without a license key. After 30 days, you must purchase a connector subscription which includes [Confluent enterprise license](/platform/current/installation/license.html#enterprise-subscription-license) keys to subscribers, along with [enterprise-level support](https://www.confluent.io/subscription/) for Confluent Platform and your connectors. If you are a subscriber, you can contact [Confluent Support](https://support.confluent.io/) for more information. For license properties, see [Confluent Platform license](configuration_options.md#hdfs2-source-connector-license-config). For information about the license topic, see [Confluent License Properties](configuration_options.md#hdfs2-source-license-topic-configuration). ## Configuration Properties For a complete list of configuration properties for this connector, see [Configuration Reference for HDFS 2 Source Connector for Confluent Platform](configuration_options.md#hdfs2-source-config-options). ## Install the HDFS 2 Source Connector You can install this connector by using the [confluent connect plugin install](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html) command, or by manually downloading the ZIP file. ### Prerequisites - You must install the connector on every machine where Connect will run. - Kafka Broker: Confluent Platform 3.3.0 or later, or Kafka 0.11.0 or later. - Connect: Confluent Platform 4.0.0 or later, or Kafka 1.0.0 or later. - Java 8+. Note that Java 8 is deprecated in versions 7.2 and later of Confluent Platform. For more details, view [Java compatibility with Confluent Platform](/platform/current/installation/versions-interoperability.html#java) by version. Java 17 users must add the following `--add-opens` JVM flag to enable the Hive integration feature: ```text --add-opens java.base/java.net=ALL-UNNAMED ``` - An installation of the latest (`latest`) connector version. ### Install the connector using Confluent CLI To install the `latest` connector version, navigate to your Confluent Platform installation directory and run the following command: ```bash confluent connect plugin install confluentinc/kafka-connect-hdfs2-source:latest ``` You can install a specific version by replacing `latest` with a version number as shown in the following example: ```bash confluent connect plugin install confluentinc/kafka-connect-hdfs2-source:3.1.0 ``` ### Install the connector manually [Download and extract the ZIP file](https://www.confluent.io/hub/confluentinc/kafka-connect-hdfs2-source) for your connector and then follow the manual connector installation [instructions](/kafka-connectors/self-managed/userguide.html#connect-installing-plugins). ## Quick Start This quick start uses the HDFS 2 Source connector to export Avro data to a Kafka topic produced by the HDFS 2 Sink connector. Before you start connector, make sure Hadoop is running locally or remotely and that you know the HDFS URL. This quick start assumes that you started the required services with the default configurations and you should make necessary changes according to the actual configurations used. #### NOTE You must ensure the connector user has read access to the directories specified in `topics.dir`. The default value of `topics.dir` is `/topics`.The following uses the `Hdfs2SinkConnector` to write a file from the Kafka topic named `test_hdfs` to HDFS 2. Then, the `HDFS2SourceConnector` loads that Avro file from HDFS 2 to the Kafka topic named `copy_of_test_hdfs`. The following uses the `HDFS2SinkConnector` to write a file from the Kafka topic named `hdfs2_topic` to HDFS 2 cluster. Then, the `HDFS2SourceConnector` loads that Avro file from HDFS 2 to the Kafka topic named `copy_of_hdfs2_topic`. 1. Run the Hadoop 2 Docker image using the following command: ```bash docker run -it -p 9000:9000 sequenceiq/hadoop-docker:2.7.0 /etc/bootstrap.sh -bash ``` 2. Run the following commands in Docker bash: ```bash cd /usr/local/hadoop bin/hdfs dfsadmin -safemode leave bin/hdfs dfs -chmod 777 / ``` Follow the instructions from [HDFS 2 Sink connector for Confluent Platform](https://docs.confluent.io/kafka-connect-hdfs/current/index.html) to set up the data to use below. 1. Start Confluent Platform. ```bash confluent local start ``` ### Property-based example 1. Create a configuration file for the connector. This file is included with the connector in `etc/kafka-connect-hdfs2-source/hdfs2-source.properties`. This configuration is used typically along with [standalone workers](/platform/current/connect/concepts.html#standalone-workers). ```bash name=hdfs2-source connector.class=io.confluent.connect.hdfs2.Hdfs2SourceConnector tasks.max=1 store.url=hdfs://localhost:9000 format.class=io.confluent.connect.hdfs2.format.avro.AvroFormat confluent.topic.bootstrap.servers=localhost:9092 confluent.topic.replication.factor=1 ``` 2. Edit the `hdfs2-source.properties` to add the following properties: ```properties transforms=AddPrefix transforms.AddPrefix.type=org.apache.kafka.connect.transforms.RegexRouter transforms.AddPrefix.regex=.* transforms.AddPrefix.replacement=copy_of_$0 ``` #### IMPORTANT Adding this renames the output of topic of the messages to `copy_of_test_hdfs`. This prevents a continuous feedback loop of messages if we have both the sink and source connectors operating on the same Kafka topic. 1. Load the HDFS 2 Source connector. ```bash confluent local load hdfs2-source --config hdfs2-source.properties ``` #### IMPORTANT Don’t use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) in production environments. 2. Confirm the connector is in a `RUNNING` state. ```bash confluent local status hdfs2-source ``` 3. Validate that the Avro data is in the Kafka topic. ```bash kafka-avro-console-consumer \ --bootstrap-server localhost:9092 \ --property schema.registry.url=http://localhost:8081 \ --topic copy_of_test_hdfs \ --from-beginning | jq '.' ``` The response should be three records as shown in the following example: ```bash {"f1": "value1"} {"f1": "value2"} {"f1": "value3"} ``` ### REST-based example Use this setting with [distributed workers](/platform/current/connect/concepts.html#distributed-workers). Write the following JSON to `config.json`, configure all of the required values, and use the following command to post the configuration to one of the distributed connect workers. Check here for more information about the Kafka Connect [REST API](/platform/current/connect/references/restapi.html) ```json { "name" : "hdfs2-source", "config" : { "connector.class" : "io.confluent.connect.hdfs2.Hdfs2SourceConnector", "tasks.max" : "1", "store.url" : "hdfs://localhost:9000", "format.class" : "io.confluent.connect.hdfs2.format.avro.AvroFormat", "confluent.topic.bootstrap.servers" : "localhost:9092", "confluent.topic.replication.factor" :"1", "transforms" : "AddPrefix", "transforms.AddPrefix.type" : "org.apache.kafka.connect.transforms.RegexRouter", "transforms.AddPrefix.regex" : ".*", "transforms.AddPrefix.replacement" : "copy_of_$0" } } ``` #### NOTE Change the `confluent.topic.bootstrap.servers` property to include your broker address(es), and change the `confluent.topic.replication.factor` to 3 for staging or production use. Use curl to post a configuration to one of the Kafka Connect Workers. Change `http://localhost:8083/` to the endpoint of one of your Kafka Connect worker(s). ```bash curl -s -X POST -H 'Content-Type: application/json' --data @config.json http://localhost:8083/connectors ``` Use the following command to update the configuration of existing connector. ```bash curl -s -X PUT -H 'Content-Type: application/json' --data @config.json http://localhost:8083/connectors/HDFS2SourceConnector/config ``` To consume records written by the connector to the configured Kafka topic, run the following command: ```bash kafka-avro-console-consumer --bootstrap-server localhost:9092 --property schema.registry.url=http://localhost:8081 --topic copy_of_test_hdfs --from-beginning ``` ## HDFS 2 Source Connector Partitions The connector comes out of the box with partitioners that support default partitioning based on Kafka partitions, field partitioning, and time-based partitioning in days or hours. You may implement your own partitioners by extending the Partitioner class. The following partitioners are available by default: * **DefaultPartitioner**: To use `DefaultPartitioner` you have to configure the `partition.class`:`io.confluent.connect.storage.partitioner.DefaultPartitioner`. This partitioner helps to read the data from hadoop2 files which are of the form `//partition=/+++.` and put it in to the kafka topic. * **TimeBasedPartitioner** : The `io.confluent.connect.storage.partitioner.TimeBasedPartitioner` determines the partition from the year, month, day, hour, minutes, and/or seconds. This partitioner requires the following connector configuration properties: * The `path.format` configuration property specifies the pattern used for the `` portion of the HDFS 2 file name. For example, when `path.format='year'=YYYY/'month'=MM/'day'=dd/'hour'=HH`, it will pick Hdfs2 file names of the form `//year=YYYY/month=MM/day=dd/hour=HH/+++.`. * **HourlyPartitioner**: To use `HourlyPartitioner` you have to configure the `partitioner.class` : `io.confluent.connect.storage.partitioner.HourlyPartitioner`. The `HourlyPartitioner` is equivalent to the TimeBasedPartitioner with `path.format='year'=YYYY/'month'=MM/'day'=dd/'hour'=HH` and `partition.duration.ms=3600000`.(one hour, for one HDFS 2 file in each hourly directory). This partitioner always results in HDFS 2 file names of the form `//year=YYYY/month=MM/day=dd/hour=HH/+++.`. * **DailyPartitioner**: To use `DailyPartitioner` you have to configure the `partitioner.class` : `io.confluent.connect.storage.partitioner.DailyPartitioner`.The `DailyPartitioner` is equivalent to the TimeBasedPartitioner with `path.format='year'=YYYY/'month'=MM/'day'=dd` and `partition.duration.ms=86400000` (one day, for one HDFS 2 file in each daily directory). This partitioner will pick HDFS 2 file of the form `//year=YYYY/month=MM/day=dd/+++.`. * **FieldPartitioner**: To use `FieldPartitioner` you have to configure the `partitioner.class` : `io.confluent.connect.storage.partitioner.FieldPartitioner`.The `` is always `/=`, resulting in HDFS 2 file names of the form `//=/+++.`. This partitioner requires the following connector configuration properties: * The `partition.field.name` configuration property specifies the pattern of the directory structure used for the `` portion of the HDFS 2 file name. ## HDFS 2 Source Connector Data Formats The HDFS 2 Source connector supports the following data formats: * **Avro Format** : For supporting Avro Format. You must configure the `format.class` = `io.confluent.connect.hdfs2.format.avro.AvroFormat` * **Json Format** : For supporting Json Format. You must configure the `format.class` = `io.confluent.connect.hdfs2.format.json.JsonFormat` * **String Format** : For supporting String Format. You must configure the `format.class` = `io.confluent.connect.hdfs2.format.string.StringFormat` * **Parquet Format** : For supporting Parquet Format. You must configure the `format.class` = `io.confluent.connect.hdfs2.format.parquet.ParquetFormat` ## Secure HDFS with Kerberos The connector supports Kerberos authentication to support secure HDFS. To work with secure HDFS, you need to specify `hdfs.authentication.kerberos`, `connect.hdfs.principal`, `connect.keytab`, `hdfs.namenode.principal`. ```properties hdfs.authentication.kerberos=true connect.hdfs.principal=connect-hdfs/_HOST@YOUR-REALM.COM connect.hdfs.keytab=/full/path/to/the/connector/keytab hdfs.namenode.principal=namenode-principal ``` You need to create the Kafka connect principals and keytab files using Kerberos and distribute the keytab file to all hosts running the connector. Make sure that only the connector user has read access to the keytab file. Currently, the connector requires that the principal and the keytab path to be the same on all the hosts running the connector. ## Troubleshooting Connector and Task Failures ### Stack trace You can use the Connect [REST API](/platform/current/connect/references/restapi.html) to check the status of the connectors and tasks. If a task or connector has failed, the `trace` field will include a reason and a stack trace. ### Fewer partitions in destination cluster If there are fewer partitions in the destination cluster than in the source topic, the connector task throws an exception and immediately stops when it tries to write to a Kafka partition that does not exist. You will see the following error messages in the Connect worker log. The recommended practice is to create topics manually in the destination Kafka cluster with the correct number of partitions before running the source connector. ```text INFO WorkerSourceTask{id=hdfs2-source-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask:409) INFO WorkerSourceTask{id=hdfs2-source-0} flushing 1 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:426) ERROR WorkerSourceTask{id=hdfs2-source-0} Failed to flush, timed out while waiting for producer to flush outstanding 1 messages (org.apache.kafka.connect.runtime.WorkerSourceTask:431) ERROR WorkerSourceTask{id=hdfs2-source-0} Failed to commit offsets (org.apache.kafka.connect.runtime.SourceTaskOffsetCommitter:114) ``` #### Error Handling The following `behavior.on.error` configuration properties set how the connector handles errors. * `fail` : The connector stops processing when an error occurs. The full batch of records will not be sent to Kafka if any record in the batch is corrupted. * `ignore` : The corrupted record is ignored. The connector continues processing the next record. For Avro, the connector ignores the file containing a corrupted record and continues processing records for the next file. * `log` : Logs an error message and continues processing the next record. For Avro, the connector ignores the file containing a corrupted record and continues processing records for the next file. For Parquet Format, when `behavior.on.error` sets to `log` or `ignore`, the connector ignores the file containing a corrupted record and continues processing records for the next file. #### NOTE The connector always ignores a file which is not in `+++.` format.