Repository: dragonflyoss/Dragonfly2
Pull request #420 (2021-07-04 04:44:03)

feat: CDN implementation supports HDFS type storage
<!--- Provide a general summary of your changes in the Title above -->

## Description

CDN implementation supports HDFS type storage

## Related Issue

https://github.com/dragonflyoss/Dragonfly2/issues/384

## Motivation and Context

<!--- Why is this change required? What problem does it solve? -->


## Screenshots (if appropriate):

1、User this HDFS client；[https://github.com/colinmarc/hdfs](https://github.com/colinmarc/hdfs)
2、Local Hadoop cluster by this repo implement；[https://github.com/big-data-europe/docker-hadoop](https://github.com/big-data-europe/docker-hadoop)

## Types of changes

<!--- What types of changes does your code introduce? Put an `x` in all the boxes that apply: -->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to change)
- [ ] Documentation Update (if none of the other choices apply)

## Checklist:

<!--- Go over all the following points, and put an `x` in all the boxes that apply. -->
<!--- If you're unsure about any of these, don't hesitate to ask. We're here to help! -->

- [x] Code compiles correctly.
- [x] My code follows the code style of this project.
- [ ] My change requires a change to the documentation.
- [ ] I have updated the documentation accordingly.
- [x] I have read the **CONTRIBUTING** document.
- [x] I have added tests to cover my changes.
- [x] All new and existing tests passed.


codecov-commenter: # [Codecov](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/420?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#420](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/420?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (b377d35) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/54c4bf8fb20000222a15fe6132b9e540d4438f64?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (54c4bf8) will **increase** coverage by `0.46%`.
> The diff coverage is `47.16%`.

[![Impacted file tree graph](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/420/graphs/tree.svg?width=650&height=150&src=pr&token=PUICJYSAT7&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/420?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)

```diff
@@            Coverage Diff             @@
##             main     #420      +/-   ##
==========================================
+ Coverage   38.24%   38.71%   +0.46%     
==========================================
  Files          80       81       +1     
  Lines        6212     6321     +109     
==========================================
+ Hits         2376     2447      +71     
- Misses       3500     3519      +19     
- Partials      336      355      +19     
```

| Flag | Coverage Δ | |
|---|---|---|
| unittests | `38.71% <47.16%> (+0.46%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/420?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [pkg/source/headers.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/420/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-cGtnL3NvdXJjZS9oZWFkZXJzLmdv) | `0.00% <0.00%> (ø)` | |
| [pkg/source/hdfsprotocol/hdfs\_source\_client.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/420/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-cGtnL3NvdXJjZS9oZGZzcHJvdG9jb2wvaGRmc19zb3VyY2VfY2xpZW50Lmdv) | `50.00% <48.54%> (ø)` | |
| [client/daemon/peer/peertask\_base.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/420/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX2Jhc2UuZ28=) | `33.49% <0.00%> (-0.98%)` | :arrow_down: |
| [client/daemon/peer/peertask\_stream.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/420/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3N0cmVhbS5nbw==) | `58.11% <0.00%> (+2.26%)` | :arrow_up: |
| [client/daemon/peer/peertask\_stream\_callback.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/420/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3N0cmVhbV9jYWxsYmFjay5nbw==) | `65.00% <0.00%> (+20.00%)` | :arrow_up: |


Patch:
diff --git a/Makefile b/Makefile
index d2665c91..a1da75ae 100644
--- a/Makefile
+++ b/Makefile
@@ -181,12 +181,12 @@ build-dfget-man-page:
 
 # Run unittests
 test:
-	@go test -race -short ${PKG_LIST}
+	@go test -gcflags "all=-l" -race -short ${PKG_LIST}
 .PHONY: test
 
 # Run tests with coverage
 test-coverage:
-	@go test -race -short ${PKG_LIST} -coverprofile cover.out -covermode=atomic
+	@go test -gcflags "all=-l" -race -short ${PKG_LIST} -coverprofile cover.out -covermode=atomic
 	@cat cover.out >> coverage.txt
 .PHONY: test-coverage
 
@@ -195,12 +195,12 @@ install-actions-e2e-test:
 	@./hack/install-e2e-test.sh actions
 .PHONY: install-actions-e2e-test
 
-# Run github actons E2E tests
+# Run github actions E2E tests
 actions-e2e-test: install-actions-e2e-test
 	@ginkgo -v -r --failFast test/e2e --trace --progress
 .PHONY: actions-e2e-test
 
-# Run github actons E2E tests with coverage
+# Run github actions E2E tests with coverage
 actions-e2e-test-coverage: install-actions-e2e-test
 	@ginkgo -v -r --failFast -cover test/e2e --trace --progress
 	@cat test/e2e/*.coverprofile >> coverage.txt
diff --git a/go.mod b/go.mod
index e158739e..036693be 100644
--- a/go.mod
+++ b/go.mod
@@ -5,9 +5,11 @@ go 1.15
 require (
 	github.com/HuKeping/rbtree v0.0.0-20210106022122-8ad34838eb2b
 	github.com/VividCortex/mysqlerr v1.0.0
+	github.com/agiledragon/gomonkey v2.0.2+incompatible
 	github.com/aliyun/aliyun-oss-go-sdk v2.1.6+incompatible
 	github.com/appleboy/gin-jwt/v2 v2.6.4
 	github.com/baiyubin/aliyun-sts-go-sdk v0.0.0-20180326062324-cfa1a18b161f // indirect
+	github.com/colinmarc/hdfs/v2 v2.2.0
 	github.com/docker/go-units v0.4.0
 	github.com/emirpasic/gods v1.12.0
 	github.com/envoyproxy/protoc-gen-validate v0.6.1
diff --git a/go.sum b/go.sum
index 507659f6..deeb7f40 100644
--- a/go.sum
+++ b/go.sum
@@ -23,6 +23,8 @@ github.com/PuerkitoBio/purell v1.1.1/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbt
 github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578/go.mod h1:uGdkoq3SwY9Y+13GIhn11/XLaGBb4BfwItxLd5jeuXE=
 github.com/VividCortex/mysqlerr v1.0.0 h1:5pZ2TZA+YnzPgzBfiUWGqWmKDVNBdrkf9g+DNe1Tiq8=
 github.com/VividCortex/mysqlerr v1.0.0/go.mod h1:xERx8E4tBhLvpjzdUyQiSfUxeMcATEQrflDAfXsqcAE=
+github.com/agiledragon/gomonkey v2.0.2+incompatible h1:eXKi9/piiC3cjJD1658mEE2o3NjkJ5vDLgYjCQu0Xlw=
+github.com/agiledragon/gomonkey v2.0.2+incompatible/go.mod h1:2NGfXu1a80LLr2cmWXGBDaHEjb1idR6+FVlX5T3D9hw=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
@@ -54,6 +56,8 @@ github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDk
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/cockroachdb/apd v1.1.0 h1:3LFP3629v+1aKXU5Q37mxmRxX/pIu1nijXydLShEq5I=
 github.com/cockroachdb/apd v1.1.0/go.mod h1:8Sl8LxpKi29FqWXR16WEFZRNSz3SoPzUzeMeY4+DwBQ=
+github.com/colinmarc/hdfs/v2 v2.2.0 h1:4AaIlTq+/sWmeqYhI0dX8bD4YrMQM990tRjm636FkGM=
+github.com/colinmarc/hdfs/v2 v2.2.0/go.mod h1:Wss6n3mtaZyRwWaqtSH+6ge01qT0rw9dJJmvoUnIQ/E=
 github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
 github.com/coreos/etcd v3.3.13+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
@@ -156,6 +160,7 @@ github.com/golang/mock v1.2.0/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfb
 github.com/golang/mock v1.3.1/go.mod h1:sBzyDLLjw3U8JLTeZvSv8jJB+tU5PVekmnlKIyFUx0Y=
 github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
 github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs=
+github.com/golang/protobuf v1.1.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
@@ -202,6 +207,10 @@ github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 h1:EGx4pi6eqNxGa
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
 github.com/gorilla/mux v1.7.3 h1:gnP5JzjVOuiZD07fKKToCAOjS0yOpj/qPETTXCCS6hw=
 github.com/gorilla/mux v1.7.3/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs=
+github.com/gorilla/securecookie v1.1.1 h1:miw7JPhV+b/lAHSXz4qd/nN9jRiAFV5FwjeKyCS8BvQ=
+github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+HVt/4epWDjd4=
+github.com/gorilla/sessions v1.2.0 h1:S7P+1Hm5V/AT9cjEcUD5uDaQSX0OE577aCXgoaKpYbQ=
+github.com/gorilla/sessions v1.2.0/go.mod h1:dk2InVEVJ0sfLlnXv9EAgkf6ecYs/i80K/zI+bUmuGM=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk=
@@ -218,6 +227,8 @@ github.com/hashicorp/go-sockaddr v1.0.0/go.mod h1:7Xibr9yA9JjQq1JpNB2Vw7kxv8xerX
 github.com/hashicorp/go-syslog v1.0.0/go.mod h1:qPfqrKkXGihmCqbJM2mZgkZGvKG1dFdvsLplgctolz4=
 github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
 github.com/hashicorp/go-uuid v1.0.1/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
+github.com/hashicorp/go-uuid v1.0.2 h1:cfejS+Tpcp13yd5nYHWDI6qVCny6wyX2Mt5SGur2IGE=
+github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
 github.com/hashicorp/go.net v0.0.1/go.mod h1:hjKkEWcCURg++eb33jQU7oqQcI9XDCnUzHA0oac0k90=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
@@ -285,6 +296,18 @@ github.com/jackc/puddle v1.1.1/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dv
 github.com/jackc/puddle v1.1.3/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/jarcoal/httpmock v1.0.8 h1:8kI16SoO6LQKgPE7PvQuV+YuD/inwHd7fOOe2zMbo4k=
 github.com/jarcoal/httpmock v1.0.8/go.mod h1:ATjnClrvW/3tijVmpL/va5Z3aAyGvqU3gCT8nX0Txik=
+github.com/jcmturner/aescts/v2 v2.0.0 h1:9YKLH6ey7H4eDBXW8khjYslgyqG2xZikXP0EQFKrle8=
+github.com/jcmturner/aescts/v2 v2.0.0/go.mod h1:AiaICIRyfYg35RUkr8yESTqvSy7csK90qZ5xfvvsoNs=
+github.com/jcmturner/dnsutils/v2 v2.0.0 h1:lltnkeZGL0wILNvrNiVCR6Ro5PGU/SeBvVO/8c/iPbo=
+github.com/jcmturner/dnsutils/v2 v2.0.0/go.mod h1:b0TnjGOvI/n42bZa+hmXL+kFJZsFT7G4t3HTlQ184QM=
+github.com/jcmturner/gofork v1.0.0 h1:J7uCkflzTEhUZ64xqKnkDxq3kzc96ajM1Gli5ktUem8=
+github.com/jcmturner/gofork v1.0.0/go.mod h1:MK8+TM0La+2rjBD4jE12Kj1pCCxK7d2LK/UM3ncEo0o=
+github.com/jcmturner/goidentity/v6 v6.0.1 h1:VKnZd2oEIMorCTsFBnJWbExfNN7yZr3EhJAxwOkZg6o=
+github.com/jcmturner/goidentity/v6 v6.0.1/go.mod h1:X1YW3bgtvwAXju7V3LCIMpY0Gbxyjn/mY9zx4tFonSg=
+github.com/jcmturner/gokrb5/v8 v8.4.1 h1:IGSJfqBzMS6TA0oJ7DxXdyzPK563QHa8T2IqER2ggyQ=
+github.com/jcmturner/gokrb5/v8 v8.4.1/go.mod h1:T1hnNppQsBtxW0tCHMHTkAt8n/sABdzZgZdoFrZaZNM=
+github.com/jcmturner/rpc/v2 v2.0.2 h1:gMB4IwRXYsWw4Bc6o/az2HJgFUA1ffSh90i26ZJ6Xl0=
+github.com/jcmturner/rpc/v2 v2.0.2/go.mod h1:VUJYCIDm3PVOEHw8sgt091/20OJjskO/YJki3ELg/Hc=
 github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E=
 github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc=
 github.com/jinzhu/now v1.1.1/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
@@ -400,6 +423,7 @@ github.com/onsi/gomega v1.10.5/go.mod h1:gza4q3jKQJijlu05nKWRCW/GavJumGt8aNRxWg7
 github.com/onsi/gomega v1.14.0 h1:ep6kpPVwmr/nTbklSx2nrLNSIO62DoYAhnPNIMhK8gI=
 github.com/onsi/gomega v1.14.0/go.mod h1:cIuvLEne0aoVhAgh/O6ac0Op8WWw9H6eYCriF+tEHG0=
 github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
+github.com/pborman/getopt v0.0.0-20180729010549-6fdd0a2c7117/go.mod h1:85jBQOZwpVEaDAr341tbn15RS4fCAsIst0qp7i8ex1o=
 github.com/pborman/uuid v1.2.1 h1:+ZZIw58t/ozdjRaXh/3awHfmWRbzYxJoAdNJxe/3pvw=
 github.com/pborman/uuid v1.2.1/go.mod h1:X/NO0urCmaxf9VXbdlT7C2Yzkj2IKimNn4k+gtPdI/k=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
@@ -568,6 +592,7 @@ golang.org/x/crypto v0.0.0-20190611184440-5c40567a22f8/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20190820162420-60c769a6c586/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190911031432-227b76d455e7/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20200117160349-530e935923ad/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
@@ -622,6 +647,7 @@ golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190827160401-ba9fcec4b297/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200324143707-d3edc9973b7e/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
diff --git a/pkg/source/hdfsprotocol/hdfs_source_client.go b/pkg/source/hdfsprotocol/hdfs_source_client.go
index ecdcafc3..91da5082 100644
--- a/pkg/source/hdfsprotocol/hdfs_source_client.go
+++ b/pkg/source/hdfsprotocol/hdfs_source_client.go
@@ -17,15 +17,34 @@
 package hdfsprotocol
 
 import (
+	"bytes"
 	"context"
 	"io"
+	"net/url"
+	"os/user"
+	"strings"
+	"sync"
+	"time"
+
+	"d7y.io/dragonfly/v2/pkg/util/rangeutils"
+
+	"github.com/go-http-utils/headers"
+	"github.com/pkg/errors"
 
 	"d7y.io/dragonfly/v2/pkg/source"
+	"github.com/colinmarc/hdfs/v2"
 )
 
 const (
 	HDFSClient = "hdfs"
 )
+const (
+	layout = "2006-01-02 15:04:05"
+	// hdfsUseDataNodeHostName set hdfs client whether user hostname connect to datanode
+	hdfsUseDataNodeHostName = "dfs.client.use.datanode.hostname"
+	// hdfsUseDataNodeHostNameValue set value is true
+	hdfsUseDataNodeHostNameValue = "true"
+)
 
 func init() {
 	source.Register(HDFSClient, NewHDFSSourceClient())
@@ -33,36 +52,222 @@ func init() {
 
 // hdfsSourceClient is an implementation of the interface of SourceClient.
 type hdfsSourceClient struct {
+	sync.RWMutex
+	clientMap map[string]*hdfs.Client
+}
+
+// hdfsFileReaderClose is an combination object of the  io.LimitedReader and io.Closer
+type hdfsFileReaderClose struct {
+	limited io.Reader
+	c       io.Closer
+	buf     *bytes.Buffer
+}
+
+func newHdfsFileReaderClose(r io.Reader, n int64, c io.Closer) io.ReadCloser {
+	return &hdfsFileReaderClose{
+		limited: io.LimitReader(r, n),
+		c:       c,
+		buf:     bytes.NewBuffer(make([]byte, 512)),
+	}
 }
 
 type HDFSSourceClientOption func(p *hdfsSourceClient)
 
-func (h hdfsSourceClient) GetContentLength(ctx context.Context, url string, header source.RequestHeader) (int64, error) {
-	panic("implement me")
+func (h *hdfsSourceClient) GetContentLength(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (int64, error) {
+	hdfsClient, path, err := h.getHDFSClientAndPath(url)
+	if err != nil {
+		return -1, err
+	}
+	info, err := hdfsClient.Stat(path)
+	if err != nil {
+		return -1, err
+	}
+
+	if rang != nil {
+
+		if int64(rang.EndIndex) <= info.Size() {
+			return int64(rang.EndIndex - rang.StartIndex), nil
+		}
+		return info.Size() - int64(rang.StartIndex), nil
+	}
+
+	return info.Size(), nil
 }
 
-func (h hdfsSourceClient) IsSupportRange(ctx context.Context, url string, header source.RequestHeader) (bool, error) {
-	panic("implement me")
+func (h *hdfsSourceClient) IsSupportRange(ctx context.Context, url string, header source.RequestHeader) (bool, error) {
+	hdfsClient, path, err := h.getHDFSClientAndPath(url)
+	if err != nil {
+		return false, err
+	}
+	_, err = hdfsClient.Stat(path)
+	if err != nil {
+		return false, err
+	}
+	return true, nil
 }
 
-func (h hdfsSourceClient) IsExpired(ctx context.Context, url string, header source.RequestHeader, expireInfo map[string]string) (bool, error) {
-	panic("implement me")
+func (h *hdfsSourceClient) IsExpired(ctx context.Context, url string, header source.RequestHeader, expireInfo map[string]string) (bool, error) {
+	lastModified := expireInfo[headers.LastModified]
+	//eTag := expireInfo[headers.ETag]
+	if lastModified == "" {
+		return true, nil
+	}
+
+	hdfsClient, path, err := h.getHDFSClientAndPath(url)
+	if err != nil {
+		return false, err
+	}
+
+	info, err := hdfsClient.Stat(path)
+	if err != nil {
+		return false, err
+	}
+	t, err := time.ParseInLocation(layout, lastModified, time.Local)
+	if err != nil {
+		return false, err
+	}
+	return info.ModTime().Format(layout) != t.Format(layout), nil
 }
 
-func (h hdfsSourceClient) Download(ctx context.Context, url string, header source.RequestHeader) (io.ReadCloser, error) {
-	panic("implement me")
+func (h *hdfsSourceClient) Download(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, error) {
+	hdfsClient, path, err := h.getHDFSClientAndPath(url)
+	if err != nil {
+		return nil, err
+	}
+	hdfsFile, err := hdfsClient.Open(path)
+	if err != nil {
+		return nil, err
+	}
+
+	// default read all data when rang is nil
+	var limitReadN int64 = hdfsFile.Stat().Size()
+
+	if rang != nil {
+		_, err = hdfsFile.Seek(int64(rang.StartIndex), 0)
+		if err != nil {
+			hdfsFile.Close()
+			return nil, err
+		}
+		limitReadN = int64(rang.EndIndex - rang.StartIndex)
+	}
+
+	return newHdfsFileReaderClose(hdfsFile, limitReadN, hdfsFile), nil
 }
 
-func (h hdfsSourceClient) DownloadWithResponseHeader(ctx context.Context, url string, header source.RequestHeader) (io.ReadCloser, source.ResponseHeader, error) {
-	panic("implement me")
+func (h *hdfsSourceClient) DownloadWithResponseHeader(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, source.ResponseHeader, error) {
+
+	hdfsClient, path, err := h.getHDFSClientAndPath(url)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	hdfsFile, err := hdfsClient.Open(path)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	fileInfo := hdfsFile.Stat()
+
+	// default read all data when rang is nil
+	var limitReadN int64 = fileInfo.Size()
+
+	if rang != nil {
+		_, err = hdfsFile.Seek(int64(rang.StartIndex), 0)
+		if err != nil {
+			hdfsFile.Close()
+			return nil, nil, err
+		}
+		limitReadN = int64(rang.EndIndex - rang.StartIndex)
+	}
+
+	return newHdfsFileReaderClose(hdfsFile, limitReadN, hdfsFile), source.ResponseHeader{
+		source.LastModified: fileInfo.ModTime().Format(layout),
+	}, nil
+}
+
+func (h *hdfsSourceClient) GetLastModifiedMillis(ctx context.Context, url string, header source.RequestHeader) (int64, error) {
+
+	hdfsClient, path, err := h.getHDFSClientAndPath(url)
+	if err != nil {
+		return -1, err
+	}
+
+	info, err := hdfsClient.Stat(path)
+	if err != nil {
+		return -1, err
+	}
+
+	return info.ModTime().UnixNano() / time.Millisecond.Nanoseconds(), nil
+}
+
+// getHDFSClient return hdfs client
+func (h *hdfsSourceClient) getHDFSClient(rawurl string) (*hdfs.Client, error) {
+	if len(rawurl) < 4 {
+		return nil, errors.Errorf("hdfs url invalid: url is %s", rawurl)
+	}
+
+	parse, err := url.Parse(rawurl)
+	if err != nil {
+		return nil, err
+	}
+
+	// get client for map
+	h.RWMutex.RLock()
+	if client, ok := h.clientMap[parse.Host]; ok {
+		h.RWMutex.RUnlock()
+		return client, nil
+	}
+	h.RWMutex.RUnlock()
+
+	// create client option
+	options := hdfs.ClientOptionsFromConf(map[string]string{
+		hdfsUseDataNodeHostName: hdfsUseDataNodeHostNameValue,
+	})
+	options.Addresses = strings.Split(parse.Host, ",")
+	u, err := user.Current()
+	if err != nil {
+		return nil, err
+	}
+	options.User = u.Username
+
+	// create hdfs client and put map
+	h.RWMutex.Lock()
+	client, err := hdfs.NewClient(options)
+	if err != nil {
+		h.RWMutex.Unlock()
+		return nil, err
+	}
+	h.clientMap[parse.Host] = client
+	h.RWMutex.Unlock()
+	return client, err
 }
 
-func (h hdfsSourceClient) GetLastModifiedMillis(ctx context.Context, url string, header source.RequestHeader) (int64, error) {
-	panic("implement me")
+// getHDFSPath return file path
+func (h *hdfsSourceClient) getHDFSPath(urls string) (string, error) {
+	parse, err := url.Parse(urls)
+	if err != nil {
+		return "", err
+	}
+	return parse.Path, nil
+}
+
+// getHDFSClientAndPath return client and path
+func (h *hdfsSourceClient) getHDFSClientAndPath(urls string) (*hdfs.Client, string, error) {
+	client, err := h.getHDFSClient(urls)
+	if err != nil {
+		return nil, "", errors.Errorf("hdfs create client fail, url is %s", urls)
+	}
+	path, err := h.getHDFSPath(urls)
+	if err != nil {
+		return client, "", errors.Errorf("hdfs url path parse fail, url is %s", urls)
+	}
+	return client, path, nil
 }
 
 func NewHDFSSourceClient(opts ...HDFSSourceClientOption) source.ResourceClient {
-	sourceClient := &hdfsSourceClient{}
+	sourceClient := &hdfsSourceClient{
+		clientMap: make(map[string]*hdfs.Client),
+	}
 	for i := range opts {
 		opts[i](sourceClient)
 	}
@@ -70,3 +275,19 @@ func NewHDFSSourceClient(opts ...HDFSSourceClientOption) source.ResourceClient {
 }
 
 var _ source.ResourceClient = (*hdfsSourceClient)(nil)
+
+func (rc *hdfsFileReaderClose) Read(p []byte) (n int, err error) {
+	return rc.limited.Read(p)
+}
+
+func (rc *hdfsFileReaderClose) Close() error {
+	return rc.c.Close()
+}
+
+func (rc *hdfsFileReaderClose) WriteTo(w io.Writer) (n int64, err error) {
+	_, err = rc.limited.Read(rc.buf.Bytes())
+	if err != nil {
+		return -1, err
+	}
+	return rc.buf.WriteTo(w)
+}
diff --git a/pkg/source/headers.go b/pkg/source/headers.go
index 17d48b56..76ffdc96 100644
--- a/pkg/source/headers.go
+++ b/pkg/source/headers.go
@@ -16,10 +16,17 @@
 
 package source
 
-// request header
+// RequestHeader request header
 type RequestHeader map[string]string
 
-// response header
+func (h RequestHeader) Get(key string) string {
+	if h == nil {
+		return ""
+	}
+	return h[key]
+}
+
+// ResponseHeader response header
 type ResponseHeader map[string]string
 
 func (h ResponseHeader) Get(key string) string {
diff --git a/pkg/source/httpprotocol/http_source_client.go b/pkg/source/httpprotocol/http_source_client.go
index 4a4f5fe8..b6f9d6ef 100644
--- a/pkg/source/httpprotocol/http_source_client.go
+++ b/pkg/source/httpprotocol/http_source_client.go
@@ -24,6 +24,8 @@ import (
 	"net/http"
 	"time"
 
+	"d7y.io/dragonfly/v2/pkg/util/rangeutils"
+
 	"d7y.io/dragonfly/v2/cdnsystem/daemon/task"
 	"d7y.io/dragonfly/v2/pkg/source"
 	"d7y.io/dragonfly/v2/pkg/structure/maputils"
@@ -78,7 +80,7 @@ func WithHTTPClient(client *http.Client) HTTPSourceClientOption {
 	}
 }
 
-func (client *httpSourceClient) GetContentLength(ctx context.Context, url string, header source.RequestHeader) (int64, error) {
+func (client *httpSourceClient) GetContentLength(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (int64, error) {
 	resp, err := client.doRequest(ctx, http.MethodGet, url, header)
 	if err != nil {
 		return -1, err
@@ -132,7 +134,7 @@ func (client *httpSourceClient) IsExpired(ctx context.Context, url string, heade
 	return resp.StatusCode != http.StatusNotModified, nil
 }
 
-func (client *httpSourceClient) Download(ctx context.Context, url string, header source.RequestHeader) (io.ReadCloser, error) {
+func (client *httpSourceClient) Download(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, error) {
 	resp, err := client.doRequest(ctx, http.MethodGet, url, header)
 	if err != nil {
 		return nil, err
@@ -144,7 +146,7 @@ func (client *httpSourceClient) Download(ctx context.Context, url string, header
 	return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
 }
 
-func (client *httpSourceClient) DownloadWithResponseHeader(ctx context.Context, url string, header source.RequestHeader) (io.ReadCloser, source.ResponseHeader,
+func (client *httpSourceClient) DownloadWithResponseHeader(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, source.ResponseHeader,
 	error) {
 	resp, err := client.doRequest(ctx, http.MethodGet, url, header)
 	if err != nil {
diff --git a/pkg/source/mock/mock_source_client.go b/pkg/source/mock/mock_source_client.go
index ad35cf4c..0912f60e 100644
--- a/pkg/source/mock/mock_source_client.go
+++ b/pkg/source/mock/mock_source_client.go
@@ -10,6 +10,7 @@ import (
 	reflect "reflect"
 
 	source "d7y.io/dragonfly/v2/pkg/source"
+	rangeutils "d7y.io/dragonfly/v2/pkg/util/rangeutils"
 	gomock "github.com/golang/mock/gomock"
 )
 
@@ -37,24 +38,24 @@ func (m *MockResourceClient) EXPECT() *MockResourceClientMockRecorder {
 }
 
 // Download mocks base method.
-func (m *MockResourceClient) Download(arg0 context.Context, arg1 string, arg2 source.RequestHeader) (io.ReadCloser, error) {
+func (m *MockResourceClient) Download(arg0 context.Context, arg1 string, arg2 source.RequestHeader, arg3 *rangeutils.Range) (io.ReadCloser, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "Download", arg0, arg1, arg2)
+	ret := m.ctrl.Call(m, "Download", arg0, arg1, arg2, arg3)
 	ret0, _ := ret[0].(io.ReadCloser)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
 
 // Download indicates an expected call of Download.
-func (mr *MockResourceClientMockRecorder) Download(arg0, arg1, arg2 interface{}) *gomock.Call {
+func (mr *MockResourceClientMockRecorder) Download(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Download", reflect.TypeOf((*MockResourceClient)(nil).Download), arg0, arg1, arg2)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Download", reflect.TypeOf((*MockResourceClient)(nil).Download), arg0, arg1, arg2, arg3)
 }
 
 // DownloadWithResponseHeader mocks base method.
-func (m *MockResourceClient) DownloadWithResponseHeader(arg0 context.Context, arg1 string, arg2 source.RequestHeader) (io.ReadCloser, source.ResponseHeader, error) {
+func (m *MockResourceClient) DownloadWithResponseHeader(arg0 context.Context, arg1 string, arg2 source.RequestHeader, arg3 *rangeutils.Range) (io.ReadCloser, source.ResponseHeader, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "DownloadWithResponseHeader", arg0, arg1, arg2)
+	ret := m.ctrl.Call(m, "DownloadWithResponseHeader", arg0, arg1, arg2, arg3)
 	ret0, _ := ret[0].(io.ReadCloser)
 	ret1, _ := ret[1].(source.ResponseHeader)
 	ret2, _ := ret[2].(error)
@@ -62,24 +63,24 @@ func (m *MockResourceClient) DownloadWithResponseHeader(arg0 context.Context, ar
 }
 
 // DownloadWithResponseHeader indicates an expected call of DownloadWithResponseHeader.
-func (mr *MockResourceClientMockRecorder) DownloadWithResponseHeader(arg0, arg1, arg2 interface{}) *gomock.Call {
+func (mr *MockResourceClientMockRecorder) DownloadWithResponseHeader(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DownloadWithResponseHeader", reflect.TypeOf((*MockResourceClient)(nil).DownloadWithResponseHeader), arg0, arg1, arg2)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DownloadWithResponseHeader", reflect.TypeOf((*MockResourceClient)(nil).DownloadWithResponseHeader), arg0, arg1, arg2, arg3)
 }
 
 // GetContentLength mocks base method.
-func (m *MockResourceClient) GetContentLength(arg0 context.Context, arg1 string, arg2 source.RequestHeader) (int64, error) {
+func (m *MockResourceClient) GetContentLength(arg0 context.Context, arg1 string, arg2 source.RequestHeader, arg3 *rangeutils.Range) (int64, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetContentLength", arg0, arg1, arg2)
+	ret := m.ctrl.Call(m, "GetContentLength", arg0, arg1, arg2, arg3)
 	ret0, _ := ret[0].(int64)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
 
 // GetContentLength indicates an expected call of GetContentLength.
-func (mr *MockResourceClientMockRecorder) GetContentLength(arg0, arg1, arg2 interface{}) *gomock.Call {
+func (mr *MockResourceClientMockRecorder) GetContentLength(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetContentLength", reflect.TypeOf((*MockResourceClient)(nil).GetContentLength), arg0, arg1, arg2)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetContentLength", reflect.TypeOf((*MockResourceClient)(nil).GetContentLength), arg0, arg1, arg2, arg3)
 }
 
 // GetLastModifiedMillis mocks base method.
diff --git a/pkg/source/ossprotocol/oss_source_client.go b/pkg/source/ossprotocol/oss_source_client.go
index a8bd1a05..636cf3c9 100644
--- a/pkg/source/ossprotocol/oss_source_client.go
+++ b/pkg/source/ossprotocol/oss_source_client.go
@@ -25,6 +25,8 @@ import (
 	"strconv"
 	"sync"
 
+	"d7y.io/dragonfly/v2/pkg/util/rangeutils"
+
 	cdnerrors "d7y.io/dragonfly/v2/cdnsystem/errors"
 	"d7y.io/dragonfly/v2/pkg/source"
 	"d7y.io/dragonfly/v2/pkg/util/stringutils"
@@ -68,7 +70,7 @@ type ossSourceClient struct {
 	accessMap sync.Map
 }
 
-func (osc *ossSourceClient) Download(ctx context.Context, url string, header source.RequestHeader) (io.ReadCloser, error) {
+func (osc *ossSourceClient) Download(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, error) {
 	panic("implement me")
 }
 
@@ -76,7 +78,7 @@ func (osc *ossSourceClient) GetLastModifiedMillis(ctx context.Context, url strin
 	panic("implement me")
 }
 
-func (osc *ossSourceClient) GetContentLength(ctx context.Context, url string, header source.RequestHeader) (int64, error) {
+func (osc *ossSourceClient) GetContentLength(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (int64, error) {
 	resHeader, err := osc.getMeta(ctx, url, header)
 	if err != nil {
 		return -1, err
@@ -113,7 +115,7 @@ func (osc *ossSourceClient) IsExpired(ctx context.Context, url string, header so
 		HTTPHeaderEtag], nil
 }
 
-func (osc *ossSourceClient) DownloadWithResponseHeader(ctx context.Context, url string, header source.RequestHeader) (io.ReadCloser, source.ResponseHeader, error) {
+func (osc *ossSourceClient) DownloadWithResponseHeader(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, source.ResponseHeader, error) {
 	ossObject, err := parseOssObject(url)
 	if err != nil {
 		return nil, nil, errors.Wrapf(err, "parse oss object from url: %s", url)
diff --git a/pkg/source/source_client.go b/pkg/source/source_client.go
index 3152b891..460d9121 100644
--- a/pkg/source/source_client.go
+++ b/pkg/source/source_client.go
@@ -19,13 +19,17 @@ package source
 
 import (
 	"context"
-	"fmt"
 	"io"
 	"net/url"
 	"strings"
 	"sync"
 	"time"
 
+	"github.com/pkg/errors"
+
+	rangers "d7y.io/dragonfly/v2/pkg/util/rangeutils"
+	"github.com/go-http-utils/headers"
+
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 )
 
@@ -38,7 +42,7 @@ type ResourceClient interface {
 	// GetContentLength get length of resource content
 	// return -l if request fail
 	// return task.IllegalSourceFileLen if response status is not StatusOK and StatusPartialContent
-	GetContentLength(ctx context.Context, url string, header RequestHeader) (int64, error)
+	GetContentLength(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (int64, error)
 
 	// IsSupportRange checks if resource supports breakpoint continuation
 	IsSupportRange(ctx context.Context, url string, header RequestHeader) (bool, error)
@@ -47,10 +51,10 @@ type ResourceClient interface {
 	IsExpired(ctx context.Context, url string, header RequestHeader, expireInfo map[string]string) (bool, error)
 
 	// Download download from source
-	Download(ctx context.Context, url string, header RequestHeader) (io.ReadCloser, error)
+	Download(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (io.ReadCloser, error)
 
 	// DownloadWithResponseHeader download from source with responseHeader
-	DownloadWithResponseHeader(ctx context.Context, url string, header RequestHeader) (io.ReadCloser, ResponseHeader, error)
+	DownloadWithResponseHeader(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (io.ReadCloser, ResponseHeader, error)
 
 	// GetLastModified get lastModified timestamp milliseconds of resource
 	GetLastModifiedMillis(ctx context.Context, url string, header RequestHeader) (int64, error)
@@ -71,7 +75,7 @@ var _defaultMgr = &ClientManagerImpl{
 	clients: make(map[string]ResourceClient),
 }
 
-func (clientMgr *ClientManagerImpl) GetContentLength(ctx context.Context, url string, header RequestHeader) (int64, error) {
+func (clientMgr *ClientManagerImpl) GetContentLength(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (int64, error) {
 	sourceClient, err := clientMgr.getSourceClient(url)
 	if err != nil {
 		return -1, err
@@ -81,7 +85,7 @@ func (clientMgr *ClientManagerImpl) GetContentLength(ctx context.Context, url st
 		ctx, cancel = context.WithTimeout(ctx, 10*time.Second)
 		defer cancel()
 	}
-	return sourceClient.GetContentLength(ctx, url, header)
+	return sourceClient.GetContentLength(ctx, url, header, rang)
 }
 
 func (clientMgr *ClientManagerImpl) IsSupportRange(ctx context.Context, url string, header RequestHeader) (bool, error) {
@@ -110,21 +114,21 @@ func (clientMgr *ClientManagerImpl) IsExpired(ctx context.Context, url string, h
 	return sourceClient.IsExpired(ctx, url, header, expireInfo)
 }
 
-func (clientMgr *ClientManagerImpl) Download(ctx context.Context, url string, header RequestHeader) (io.ReadCloser, error) {
+func (clientMgr *ClientManagerImpl) Download(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (io.ReadCloser, error) {
 	sourceClient, err := clientMgr.getSourceClient(url)
 	if err != nil {
 		return nil, err
 	}
-	return sourceClient.Download(ctx, url, header)
+	return sourceClient.Download(ctx, url, header, rang)
 }
 
-func (clientMgr *ClientManagerImpl) DownloadWithResponseHeader(ctx context.Context, url string, header RequestHeader) (io.ReadCloser, ResponseHeader,
+func (clientMgr *ClientManagerImpl) DownloadWithResponseHeader(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (io.ReadCloser, ResponseHeader,
 	error) {
 	sourceClient, err := clientMgr.getSourceClient(url)
 	if err != nil {
 		return nil, nil, err
 	}
-	return sourceClient.DownloadWithResponseHeader(ctx, url, header)
+	return sourceClient.DownloadWithResponseHeader(ctx, url, header, rang)
 }
 
 func (clientMgr *ClientManagerImpl) GetLastModifiedMillis(ctx context.Context, url string, header RequestHeader) (int64, error) {
@@ -169,7 +173,15 @@ func UnRegister(schema string) {
 }
 
 func GetContentLength(ctx context.Context, url string, header RequestHeader) (int64, error) {
-	return _defaultMgr.GetContentLength(ctx, url, header)
+	var httpRange *rangers.Range
+	if header.Get(headers.Range) != "" {
+		var err error
+		httpRange, err = rangers.ParseHTTPRange(header.Get(headers.Range))
+		if err != nil {
+			return -1, err
+		}
+	}
+	return _defaultMgr.GetContentLength(ctx, url, header, httpRange)
 }
 
 func IsSupportRange(ctx context.Context, url string, header RequestHeader) (bool, error) {
@@ -181,11 +193,28 @@ func IsExpired(ctx context.Context, url string, header RequestHeader, expireInfo
 }
 
 func Download(ctx context.Context, url string, header RequestHeader) (io.ReadCloser, error) {
-	return _defaultMgr.Download(ctx, url, header)
+	var httpRange *rangers.Range
+	if header.Get(headers.Range) != "" {
+		var err error
+		httpRange, err = rangers.ParseHTTPRange(header.Get(headers.Range))
+		if err != nil {
+			return nil, err
+		}
+	}
+	return _defaultMgr.Download(ctx, url, header, httpRange)
 }
 
 func DownloadWithResponseHeader(ctx context.Context, url string, header RequestHeader) (io.ReadCloser, ResponseHeader, error) {
-	return _defaultMgr.DownloadWithResponseHeader(ctx, url, header)
+
+	var httpRange *rangers.Range
+	if header.Get(headers.Range) != "" {
+		var err error
+		httpRange, err = rangers.ParseHTTPRange(header.Get(headers.Range))
+		if err != nil {
+			return nil, nil, err
+		}
+	}
+	return _defaultMgr.DownloadWithResponseHeader(ctx, url, header, httpRange)
 }
 
 func GetLastModifiedMillis(ctx context.Context, url string, header RequestHeader) (int64, error) {
@@ -203,7 +232,7 @@ func (clientMgr *ClientManagerImpl) getSourceClient(rawURL string) (ResourceClie
 	client, ok := clientMgr.clients[strings.ToLower(parsedURL.Scheme)]
 	clientMgr.RUnlock()
 	if !ok || client == nil {
-		return nil, fmt.Errorf("can not find client for supporting url %s, clients: %v", rawURL, clientMgr.clients)
+		return nil, errors.Errorf("can not find client for supporting url %s, clients:%v", rawURL, clientMgr.clients)
 	}
 	return client, nil
 }


Test patch:
diff --git a/cdnsystem/daemon/cdn/manager_test.go b/cdnsystem/daemon/cdn/manager_test.go
index 73960158..1a1cc523 100644
--- a/cdnsystem/daemon/cdn/manager_test.go
+++ b/cdnsystem/daemon/cdn/manager_test.go
@@ -84,23 +84,20 @@ func (suite *CDNManagerTestSuite) TestTriggerCDN() {
 	defer source.UnRegister("http")
 	sourceClient.EXPECT().IsSupportRange(gomock.Any(), dragonflyURL, gomock.Any()).Return(true, nil).AnyTimes()
 	sourceClient.EXPECT().IsExpired(gomock.Any(), dragonflyURL, gomock.Any(), gomock.Any()).Return(false, nil).AnyTimes()
-	sourceClient.EXPECT().Download(gomock.Any(), dragonflyURL, gomock.Any()).DoAndReturn(
-		func(ctx context.Context, url string, header source.RequestHeader) (io.ReadCloser, error) {
+	sourceClient.EXPECT().Download(gomock.Any(), dragonflyURL, gomock.Any(), gomock.Any()).DoAndReturn(
+		func(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, error) {
 			content, _ := ioutil.ReadFile("../../testdata/cdn/go.html")
-			if rang, ok := header["Range"]; ok {
-				r, _ := rangeutils.ParseHTTPRange(rang)
-				return ioutil.NopCloser(io.NewSectionReader(strings.NewReader(string(content)), int64(r.StartIndex), int64(r.EndIndex))), nil
+			if rang != nil {
+				return ioutil.NopCloser(io.NewSectionReader(strings.NewReader(string(content)), int64(rang.StartIndex), int64(rang.EndIndex))), nil
 			}
 			return ioutil.NopCloser(strings.NewReader(string(content))), nil
 		},
 	).AnyTimes()
-	sourceClient.EXPECT().DownloadWithResponseHeader(gomock.Any(), dragonflyURL, gomock.Any()).DoAndReturn(
-		func(ctx context.Context, url string, header source.RequestHeader) (io.ReadCloser, source.ResponseHeader, error) {
+	sourceClient.EXPECT().DownloadWithResponseHeader(gomock.Any(), dragonflyURL, gomock.Any(), gomock.Any()).DoAndReturn(
+		func(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, source.ResponseHeader, error) {
 			content, _ := ioutil.ReadFile("../../testdata/cdn/go.html")
-			if rang, ok := header["Range"]; ok {
-				r, err := rangeutils.ParseHTTPRange(rang)
-				suite.Nil(err)
-				return ioutil.NopCloser(io.NewSectionReader(strings.NewReader(string(content)), int64(r.StartIndex), int64(r.EndIndex))),
+			if rang != nil {
+				return ioutil.NopCloser(io.NewSectionReader(strings.NewReader(string(content)), int64(rang.StartIndex), int64(rang.EndIndex))),
 					map[string]string{
 						source.LastModified: "Sun, 06 Jun 2021 12:52:30 GMT",
 						source.ETag:         "etag",
@@ -114,11 +111,10 @@ func (suite *CDNManagerTestSuite) TestTriggerCDN() {
 	).AnyTimes()
 	sourceClient.EXPECT().GetLastModifiedMillis(gomock.Any(), dragonflyURL, gomock.Any()).Return(
 		timeutils.UnixMillis("Sun, 06 Jun 2021 12:52:30 GMT"), nil).AnyTimes()
-	sourceClient.EXPECT().GetContentLength(gomock.Any(), dragonflyURL, gomock.Any()).DoAndReturn(
-		func(ctx context.Context, url string, header source.RequestHeader) (int64, error) {
-			if rang, ok := header["Range"]; ok {
-				r, _ := rangeutils.ParseHTTPRange(rang)
-				return int64(r.EndIndex-r.StartIndex) + 1, nil
+	sourceClient.EXPECT().GetContentLength(gomock.Any(), dragonflyURL, gomock.Any(), gomock.Any()).DoAndReturn(
+		func(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (int64, error) {
+			if rang != nil {
+				return int64(rang.EndIndex-rang.StartIndex) + 1, nil
 			}
 			fileInfo, _ := os.Stat("../../testdata/cdn/go.html")
 			return fileInfo.Size(), nil
diff --git a/client/daemon/peer/peertask_file_test.go b/client/daemon/peer/peertask_file_test.go
index d5148f47..93c168cf 100644
--- a/client/daemon/peer/peertask_file_test.go
+++ b/client/daemon/peer/peertask_file_test.go
@@ -26,6 +26,8 @@ import (
 	"testing"
 	"time"
 
+	rangers "d7y.io/dragonfly/v2/pkg/util/rangeutils"
+
 	"d7y.io/dragonfly/v2/client/clientutil"
 	"d7y.io/dragonfly/v2/client/config"
 	"d7y.io/dragonfly/v2/client/daemon/test"
@@ -73,12 +75,12 @@ func TestFilePeerTask_BackSource_WithContentLength(t *testing.T) {
 	sourceClient := sourceMock.NewMockResourceClient(ctrl)
 	source.Register("http", sourceClient)
 	defer source.UnRegister("http")
-	sourceClient.EXPECT().GetContentLength(gomock.Any(), url, gomock.Any()).DoAndReturn(
-		func(ctx context.Context, url string, headers source.RequestHeader) (int64, error) {
+	sourceClient.EXPECT().GetContentLength(gomock.Any(), url, gomock.Any(), gomock.Any()).DoAndReturn(
+		func(ctx context.Context, url string, headers source.RequestHeader, rang *rangers.Range) (int64, error) {
 			return int64(len(testBytes)), nil
 		})
-	sourceClient.EXPECT().Download(gomock.Any(), url, gomock.Any()).DoAndReturn(
-		func(ctx context.Context, url string, headers source.RequestHeader) (io.ReadCloser, error) {
+	sourceClient.EXPECT().Download(gomock.Any(), url, gomock.Any(), gomock.Any()).DoAndReturn(
+		func(ctx context.Context, url string, headers source.RequestHeader, rang *rangers.Range) (io.ReadCloser, error) {
 			return ioutil.NopCloser(bytes.NewBuffer(testBytes)), nil
 		})
 
@@ -183,12 +185,12 @@ func TestFilePeerTask_BackSource_WithoutContentLength(t *testing.T) {
 	sourceClient := sourceMock.NewMockResourceClient(ctrl)
 	source.Register("http", sourceClient)
 	defer source.UnRegister("http")
-	sourceClient.EXPECT().GetContentLength(gomock.Any(), url, source.RequestHeader{}).DoAndReturn(
-		func(ctx context.Context, url string, header source.RequestHeader) (int64, error) {
+	sourceClient.EXPECT().GetContentLength(gomock.Any(), url, source.RequestHeader{}, gomock.Any()).DoAndReturn(
+		func(ctx context.Context, url string, header source.RequestHeader, rang *rangers.Range) (int64, error) {
 			return -1, nil
 		})
-	sourceClient.EXPECT().Download(gomock.Any(), url, source.RequestHeader{}).DoAndReturn(
-		func(ctx context.Context, url string, header source.RequestHeader) (io.ReadCloser, error) {
+	sourceClient.EXPECT().Download(gomock.Any(), url, source.RequestHeader{}, gomock.Any()).DoAndReturn(
+		func(ctx context.Context, url string, header source.RequestHeader, rang *rangers.Range) (io.ReadCloser, error) {
 			return ioutil.NopCloser(bytes.NewBuffer(testBytes)), nil
 		})
 
diff --git a/client/daemon/peer/peertask_stream_test.go b/client/daemon/peer/peertask_stream_test.go
index 0abb40b9..31c191bb 100644
--- a/client/daemon/peer/peertask_stream_test.go
+++ b/client/daemon/peer/peertask_stream_test.go
@@ -25,6 +25,8 @@ import (
 	"testing"
 	"time"
 
+	rangers "d7y.io/dragonfly/v2/pkg/util/rangeutils"
+
 	"d7y.io/dragonfly/v2/pkg/source"
 	sourceMock "d7y.io/dragonfly/v2/pkg/source/mock"
 	"github.com/golang/mock/gomock"
@@ -71,12 +73,12 @@ func TestStreamPeerTask_BackSource_WithContentLength(t *testing.T) {
 	sourceClient := sourceMock.NewMockResourceClient(ctrl)
 	source.Register("http", sourceClient)
 	defer source.UnRegister("http")
-	sourceClient.EXPECT().GetContentLength(gomock.Any(), url, source.RequestHeader{}).DoAndReturn(
-		func(ctx context.Context, url string, headers source.RequestHeader) (int64, error) {
+	sourceClient.EXPECT().GetContentLength(gomock.Any(), url, source.RequestHeader{}, gomock.Any()).DoAndReturn(
+		func(ctx context.Context, url string, headers source.RequestHeader, rang *rangers.Range) (int64, error) {
 			return int64(len(testBytes)), nil
 		})
-	sourceClient.EXPECT().Download(gomock.Any(), url, source.RequestHeader{}).DoAndReturn(
-		func(ctx context.Context, url string, headers source.RequestHeader) (io.ReadCloser, error) {
+	sourceClient.EXPECT().Download(gomock.Any(), url, source.RequestHeader{}, gomock.Any()).DoAndReturn(
+		func(ctx context.Context, url string, headers source.RequestHeader, rang *rangers.Range) (io.ReadCloser, error) {
 			return ioutil.NopCloser(bytes.NewBuffer(testBytes)), nil
 		})
 
@@ -169,12 +171,12 @@ func TestStreamPeerTask_BackSource_WithoutContentLength(t *testing.T) {
 	sourceClient := sourceMock.NewMockResourceClient(ctrl)
 	source.Register("http", sourceClient)
 	defer source.UnRegister("http")
-	sourceClient.EXPECT().GetContentLength(gomock.Any(), url, source.RequestHeader{}).DoAndReturn(
-		func(ctx context.Context, url string, headers source.RequestHeader) (int64, error) {
+	sourceClient.EXPECT().GetContentLength(gomock.Any(), url, source.RequestHeader{}, gomock.Any()).DoAndReturn(
+		func(ctx context.Context, url string, headers source.RequestHeader, rang *rangers.Range) (int64, error) {
 			return -1, nil
 		})
-	sourceClient.EXPECT().Download(gomock.Any(), url, source.RequestHeader{}).DoAndReturn(
-		func(ctx context.Context, url string, headers source.RequestHeader) (io.ReadCloser, error) {
+	sourceClient.EXPECT().Download(gomock.Any(), url, source.RequestHeader{}, gomock.Any()).DoAndReturn(
+		func(ctx context.Context, url string, headers source.RequestHeader, rang *rangers.Range) (io.ReadCloser, error) {
 			return ioutil.NopCloser(bytes.NewBuffer(testBytes)), nil
 		})
 
diff --git a/client/dfget/dfget_test.go b/client/dfget/dfget_test.go
index 3e76bf8d..326e1e0f 100644
--- a/client/dfget/dfget_test.go
+++ b/client/dfget/dfget_test.go
@@ -52,7 +52,7 @@ func Test_downloadFromSource(t *testing.T) {
 		Digest: strings.Join([]string{constants.Sha256Hash, digestutils.Sha256(content)}, ":"),
 	}
 
-	sourceClient.EXPECT().Download(context.Background(), cfg.URL, nil).Return(ioutil.NopCloser(strings.NewReader(content)), nil)
+	sourceClient.EXPECT().Download(context.Background(), cfg.URL, nil, nil).Return(ioutil.NopCloser(strings.NewReader(content)), nil)
 
 	err = downloadFromSource(context.Background(), cfg, nil)
 	assert.Nil(t, err)
diff --git a/pkg/source/hdfsprotocol/hdfs_source_client_test.go b/pkg/source/hdfsprotocol/hdfs_source_client_test.go
new file mode 100644
index 00000000..345d4dc1
--- /dev/null
+++ b/pkg/source/hdfsprotocol/hdfs_source_client_test.go
@@ -0,0 +1,350 @@
+package hdfsprotocol
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"os"
+	"reflect"
+	"testing"
+	"time"
+
+	"d7y.io/dragonfly/v2/cdnsystem/daemon/cdn"
+	"d7y.io/dragonfly/v2/pkg/util/rangeutils"
+
+	"d7y.io/dragonfly/v2/pkg/source"
+	"github.com/agiledragon/gomonkey"
+	"github.com/colinmarc/hdfs/v2"
+	"github.com/go-http-utils/headers"
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
+)
+
+var sourceClient source.ResourceClient
+
+const (
+	hdfsExistFileHost                     = "127.0.0.1:9000"
+	hdfsExistFilePath                     = "/user/root/input/f1.txt"
+	hdfsExistFileURL                      = "hdfs://" + hdfsExistFileHost + hdfsExistFilePath
+	hdfsExistFileContentLength      int64 = 12
+	hdfsExistFileContent                  = "Hello World\n"
+	hdfsExistFileLastModifiedMillis int64 = 1625218150000
+	hdfsExistFileLastModified             = "2021-07-02 09:29:10"
+	hdfsExistFileRangeStart         int64 = 3
+	hdfsExistFileRangeEnd           int64 = 10
+)
+
+const (
+	hdfsNotExistFileURL                 = "hdfs://127.0.0.1:9000/user/root/input/f3.txt"
+	hdfsNotExistFileContentLength int64 = -1
+)
+
+var fakeHDFSClient *hdfs.Client = &hdfs.Client{}
+
+func testBefore() {
+	sourceClient = NewHDFSSourceClient(func(p *hdfsSourceClient) {
+		p.clientMap[hdfsExistFileHost] = fakeHDFSClient
+	})
+}
+
+func TestMain(t *testing.M) {
+	testBefore()
+	t.Run()
+	os.Exit(0)
+}
+
+// TestGetContentLength_OK function test exist file and return file length
+func TestGetContentLength_OK(t *testing.T) {
+
+	var info os.FileInfo = fakeHDFSFileInfo{
+		contents: hdfsExistFileContent,
+	}
+	stubRet := []gomonkey.OutputCell{
+		{Values: gomonkey.Params{info, nil}},
+	}
+
+	patch := gomonkey.ApplyMethodSeq(reflect.TypeOf(fakeHDFSClient), "Stat", stubRet)
+
+	defer patch.Reset()
+
+	// exist file
+	length, err := sourceClient.GetContentLength(context.Background(), hdfsExistFileURL, nil, &rangeutils.Range{EndIndex: 12})
+	assert.Equal(t, hdfsExistFileContentLength, length)
+	assert.Nil(t, err)
+
+}
+
+// TestGetContentLength_Fail test file not exist, return error
+func TestGetContentLength_Fail(t *testing.T) {
+
+	stubRet := []gomonkey.OutputCell{
+		{Values: gomonkey.Params{nil, errors.New("stat /user/root/input/f3.txt: file does not exist")}},
+	}
+
+	patch := gomonkey.ApplyMethodSeq(reflect.TypeOf(fakeHDFSClient), "Stat", stubRet)
+
+	defer patch.Reset()
+
+	// not exist file
+	length, err := sourceClient.GetContentLength(context.Background(), hdfsNotExistFileURL, nil, &rangeutils.Range{EndIndex: 10})
+	assert.Equal(t, hdfsNotExistFileContentLength, length)
+	assert.EqualError(t, err, "stat /user/root/input/f3.txt: file does not exist")
+}
+
+// TestIsSupportRange_FileExist test file exist, return file  support range
+func TestIsSupportRange_FileExist(t *testing.T) {
+	var info os.FileInfo = fakeHDFSFileInfo{
+		contents: hdfsExistFileContent,
+	}
+	stubRet := []gomonkey.OutputCell{
+		{Values: gomonkey.Params{info, nil}},
+	}
+
+	patch := gomonkey.ApplyMethodSeq(reflect.TypeOf(fakeHDFSClient), "Stat", stubRet)
+
+	defer patch.Reset()
+
+	supportRange, err := sourceClient.IsSupportRange(context.Background(), hdfsExistFileURL, nil)
+	assert.Equal(t, true, supportRange)
+	assert.Nil(t, err)
+}
+
+// TestIsSupportRange_FileNotExist test file not exist, return error and not support range
+func TestIsSupportRange_FileNotExist(t *testing.T) {
+	stubRet := []gomonkey.OutputCell{
+		{Values: gomonkey.Params{nil, errors.New("stat /user/root/input/f3.txt: file does not exist")}},
+	}
+
+	patch := gomonkey.ApplyMethodSeq(reflect.TypeOf(fakeHDFSClient), "Stat", stubRet)
+
+	defer patch.Reset()
+
+	supportRange, err := sourceClient.IsSupportRange(context.Background(), hdfsNotExistFileURL, nil)
+	assert.Equal(t, false, supportRange)
+	assert.EqualError(t, err, "stat /user/root/input/f3.txt: file does not exist")
+}
+
+//
+func TestIsExpired_NoHeader(t *testing.T) {
+	// header not have Last-Modified
+	expired, err := sourceClient.IsExpired(context.Background(), hdfsExistFileURL, nil, map[string]string{})
+	assert.Equal(t, true, expired)
+	assert.Nil(t, err)
+}
+func TestIsExpired_LastModifiedExpired(t *testing.T) {
+	lastModified, _ := time.Parse(layout, hdfsExistFileLastModified)
+	var info os.FileInfo = fakeHDFSFileInfo{
+		modtime: lastModified,
+	}
+	stubRet := []gomonkey.OutputCell{
+		{Values: gomonkey.Params{info, nil}},
+	}
+
+	patch := gomonkey.ApplyMethodSeq(reflect.TypeOf(fakeHDFSClient), "Stat", stubRet)
+
+	defer patch.Reset()
+
+	// header have Last-Modified
+	expired, err := sourceClient.IsExpired(context.Background(), hdfsExistFileURL, nil, map[string]string{
+		headers.LastModified: "2020-01-01 00:00:00",
+	})
+	assert.Equal(t, true, expired)
+	assert.Nil(t, err)
+
+}
+
+func TestIsExpired_LastModifiedNotExpired(t *testing.T) {
+	lastModified, _ := time.Parse(layout, hdfsExistFileLastModified)
+	var info os.FileInfo = fakeHDFSFileInfo{
+		modtime: lastModified,
+	}
+	stubRet := []gomonkey.OutputCell{
+		{Values: gomonkey.Params{info, nil}},
+	}
+
+	patch := gomonkey.ApplyMethodSeq(reflect.TypeOf(fakeHDFSClient), "Stat", stubRet)
+
+	defer patch.Reset()
+
+	// header have Last-Modified
+	expired, err := sourceClient.IsExpired(context.Background(), hdfsExistFileURL, nil, map[string]string{
+		headers.LastModified: hdfsExistFileLastModified,
+	})
+	assert.Equal(t, false, expired)
+	assert.Nil(t, err)
+}
+
+func Test_Download_FileExist_ByRang(t *testing.T) {
+	var reader *hdfs.FileReader = &hdfs.FileReader{}
+	patch := gomonkey.ApplyMethod(reflect.TypeOf(fakeHDFSClient), "Open", func(*hdfs.Client, string) (*hdfs.FileReader, error) {
+		return reader, nil
+	})
+	patch.ApplyMethod(reflect.TypeOf(reader), "Seek", func(_ *hdfs.FileReader, offset int64, whence int) (int64, error) {
+		return 0 - hdfsExistFileContentLength, nil
+	})
+	patch.ApplyMethod(reflect.TypeOf(reader), "Read", func(_ *hdfs.FileReader, b []byte) (int, error) {
+		byets := []byte(hdfsExistFileContent)
+		copy(b, byets)
+		return int(hdfsExistFileContentLength), io.EOF
+	})
+	patch.ApplyMethodSeq(reflect.TypeOf(reader), "Stat", []gomonkey.OutputCell{
+		{
+			Values: gomonkey.Params{
+				fakeHDFSFileInfo{
+					contents: hdfsExistFileContent,
+				},
+			},
+		},
+	})
+	defer patch.Reset()
+
+	rang := &rangeutils.Range{StartIndex: 0, EndIndex: uint64(hdfsExistFileContentLength)}
+	// exist file
+	download, err := sourceClient.Download(context.Background(), hdfsExistFileURL, source.RequestHeader{
+		cdn.RangeHeaderName: fmt.Sprintf("bytes=%s", rang.String()),
+	}, rang)
+	data, _ := ioutil.ReadAll(download)
+
+	assert.Equal(t, hdfsExistFileContent, string(data))
+	assert.Nil(t, err)
+
+}
+
+func TestDownload_FileNotExist(t *testing.T) {
+	stubRet := []gomonkey.OutputCell{
+		{Values: gomonkey.Params{nil, errors.New("open /user/root/input/f3.txt: file does not exist")}},
+	}
+
+	patch := gomonkey.ApplyMethodSeq(reflect.TypeOf(fakeHDFSClient), "Open", stubRet)
+
+	defer patch.Reset()
+
+	rang := rangeutils.Range{StartIndex: 0, EndIndex: uint64(hdfsExistFileContentLength)}
+	// not exist file
+	download, err := sourceClient.Download(context.Background(), hdfsNotExistFileURL, source.RequestHeader{
+		cdn.RangeHeaderName: fmt.Sprintf("bytes=%s", rang.String()),
+	}, &rang)
+	assert.Nil(t, download)
+	assert.EqualError(t, err, "open /user/root/input/f3.txt: file does not exist")
+}
+
+func Test_DownloadWithResponseHeader_FileExist_ByRange(t *testing.T) {
+	lastModified, _ := time.Parse(layout, hdfsExistFileLastModified)
+	var reader *hdfs.FileReader = &hdfs.FileReader{}
+	patches := gomonkey.NewPatches()
+	defer patches.Reset()
+	patches.ApplyMethod(reflect.TypeOf(fakeHDFSClient), "Open", func(*hdfs.Client, string) (*hdfs.FileReader, error) {
+		return reader, nil
+	})
+	patches.ApplyMethod(reflect.TypeOf(reader), "Stat", func(_ *hdfs.FileReader) os.FileInfo {
+		return fakeHDFSFileInfo{
+			contents: hdfsExistFileContent,
+			modtime:  lastModified,
+		}
+	})
+
+	patches.ApplyMethod(reflect.TypeOf(reader), "Seek", func(_ *hdfs.FileReader, offset int64, whence int) (int64, error) {
+		return hdfsExistFileRangeEnd - hdfsExistFileRangeStart, nil
+	})
+	patches.ApplyMethod(reflect.TypeOf(reader), "Read", func(_ *hdfs.FileReader, b []byte) (int, error) {
+		b = b[0 : hdfsExistFileRangeEnd-hdfsExistFileRangeStart]
+		bytes := []byte(hdfsExistFileContent)
+		copy(b, bytes[hdfsExistFileRangeStart:hdfsExistFileRangeEnd])
+		return len(b), io.EOF
+	})
+
+	rang := rangeutils.Range{StartIndex: uint64(hdfsExistFileRangeStart), EndIndex: uint64(hdfsExistFileRangeEnd)}
+	body, responseHeader, err := sourceClient.DownloadWithResponseHeader(context.Background(), hdfsExistFileURL, source.RequestHeader{
+		cdn.RangeHeaderName: fmt.Sprintf("bytes=%s", rang.String()),
+	}, &rang)
+	assert.Nil(t, err)
+	assert.Equal(t, hdfsExistFileLastModified, responseHeader.Get(source.LastModified))
+
+	data, _ := ioutil.ReadAll(body)
+	assert.Equal(t, string(data), string([]byte(hdfsExistFileContent)[hdfsExistFileRangeStart:hdfsExistFileRangeEnd]))
+}
+
+func TestDownloadWithResponseHeader_FileNotExist(t *testing.T) {
+	patch := gomonkey.ApplyMethod(reflect.TypeOf(fakeHDFSClient), "Open", func(*hdfs.Client, string) (*hdfs.FileReader, error) {
+		return nil, errors.New("open /user/root/input/f3.txt: file does not exist")
+	})
+	defer patch.Reset()
+
+	rang := rangeutils.Range{StartIndex: 0, EndIndex: uint64(hdfsExistFileContentLength)}
+	body, responseHeader, err := sourceClient.DownloadWithResponseHeader(context.Background(), hdfsNotExistFileURL, source.RequestHeader{
+		cdn.RangeHeaderName: fmt.Sprintf("bytes=%s", rang.String()),
+	}, &rang)
+	assert.EqualError(t, err, "open /user/root/input/f3.txt: file does not exist")
+	assert.Nil(t, responseHeader)
+	assert.Nil(t, body)
+}
+
+func TestGetLastModifiedMillis_FileExist(t *testing.T) {
+	lastModified, _ := time.Parse(layout, hdfsExistFileLastModified)
+	var info os.FileInfo = fakeHDFSFileInfo{
+		modtime: lastModified,
+	}
+	stubRet := []gomonkey.OutputCell{
+		{Values: gomonkey.Params{info, nil}},
+	}
+
+	patch := gomonkey.ApplyMethodSeq(reflect.TypeOf(fakeHDFSClient), "Stat", stubRet)
+
+	defer patch.Reset()
+
+	lastModifiedMillis, err := sourceClient.GetLastModifiedMillis(context.Background(), hdfsExistFileURL, nil)
+	assert.Nil(t, err)
+	assert.Equal(t, hdfsExistFileLastModifiedMillis, lastModifiedMillis)
+}
+
+func TestGetLastModifiedMillis_FileNotExist(t *testing.T) {
+	stubRet := []gomonkey.OutputCell{
+		{Values: gomonkey.Params{nil, errors.New("stat /user/root/input/f3.txt: file does not exist")}},
+	}
+
+	patch := gomonkey.ApplyMethodSeq(reflect.TypeOf(fakeHDFSClient), "Stat", stubRet)
+
+	defer patch.Reset()
+
+	lastModifiedMillis, err := sourceClient.GetLastModifiedMillis(context.Background(), hdfsNotExistFileURL, nil)
+	assert.EqualError(t, err, "stat /user/root/input/f3.txt: file does not exist")
+	assert.Equal(t, hdfsNotExistFileContentLength, lastModifiedMillis)
+}
+
+func TestNewHDFSSourceClient(t *testing.T) {
+	client := NewHDFSSourceClient()
+	assert.NotNil(t, client)
+
+	options := make([]HDFSSourceClientOption, 0)
+
+	option := func(p *hdfsSourceClient) {
+		c, _ := hdfs.New(hdfsExistFileHost)
+		p.clientMap[hdfsExistFileHost] = c
+	}
+	options = append(options, option)
+
+	newHDFSSourceClient := NewHDFSSourceClient(options...)
+
+	assert.IsType(t, &hdfsSourceClient{}, newHDFSSourceClient)
+
+}
+
+type fakeHDFSFileInfo struct {
+	dir      bool
+	basename string
+	modtime  time.Time
+	contents string
+}
+
+func (f fakeHDFSFileInfo) Name() string       { return f.basename }
+func (f fakeHDFSFileInfo) Sys() interface{}   { return nil }
+func (f fakeHDFSFileInfo) ModTime() time.Time { return f.modtime }
+func (f fakeHDFSFileInfo) IsDir() bool        { return f.dir }
+func (f fakeHDFSFileInfo) Size() int64        { return int64(len(f.contents)) }
+func (f fakeHDFSFileInfo) Mode() os.FileMode {
+	if f.dir {
+		return 0755 | os.ModeDir
+	}
+	return 0644
+}
diff --git a/pkg/source/httpprotocol/http_source_client_test.go b/pkg/source/httpprotocol/http_source_client_test.go
index 179746b3..2993749e 100644
--- a/pkg/source/httpprotocol/http_source_client_test.go
+++ b/pkg/source/httpprotocol/http_source_client_test.go
@@ -132,7 +132,7 @@ func (suite *HTTPSourceClientTestSuite) TestNewHTTPSourceClient() {
 
 func (suite *HTTPSourceClientTestSuite) TestHttpSourceClientDownloadWithResponseHeader() {
 	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
-	reader, responseHeader, err := suite.DownloadWithResponseHeader(ctx, timeoutURL, source.RequestHeader{})
+	reader, responseHeader, err := suite.DownloadWithResponseHeader(ctx, timeoutURL, source.RequestHeader{}, nil)
 	cancel()
 	suite.NotNil(err)
 	suite.Equal("Get \"http://timeout.com\": context deadline exceeded", err.Error())
@@ -205,7 +205,7 @@ func (suite *HTTPSourceClientTestSuite) TestHttpSourceClientDownloadWithResponse
 	}
 	for _, tt := range tests {
 		suite.Run(tt.name, func() {
-			reader, responseHeader, err := suite.DownloadWithResponseHeader(tt.args.ctx, tt.args.url, tt.args.header)
+			reader, responseHeader, err := suite.DownloadWithResponseHeader(tt.args.ctx, tt.args.url, tt.args.header, nil)
 			suite.Equal(tt.wantErr, err)
 			if err != nil {
 				return
@@ -238,7 +238,7 @@ func (suite *HTTPSourceClientTestSuite) TestHttpSourceClientGetContentLength() {
 	}
 	for _, tt := range tests {
 		suite.Run(tt.name, func() {
-			got, err := suite.GetContentLength(tt.args.ctx, tt.args.url, tt.args.header)
+			got, err := suite.GetContentLength(tt.args.ctx, tt.args.url, tt.args.header, nil)
 			suite.Equal(tt.wantErr, err)
 			suite.Equal(tt.want, got)
 		})
diff --git a/pkg/source/testdata/main.go b/pkg/source/testdata/main.go
index 5d5ea240..0ef6019c 100644
--- a/pkg/source/testdata/main.go
+++ b/pkg/source/testdata/main.go
@@ -42,13 +42,13 @@ func main() {
 
 	ctx := context.Background()
 
-	l, err := client.GetContentLength(ctx, "", nil)
+	l, err := client.GetContentLength(ctx, "", nil, nil)
 	if err != nil {
 		fmt.Printf("get content length error: %s\n", err)
 		os.Exit(1)
 	}
 
-	rc, err := client.Download(ctx, "", nil)
+	rc, err := client.Download(ctx, "", nil, nil)
 	if err != nil {
 		fmt.Printf("download error: %s\n", err)
 		os.Exit(1)
diff --git a/pkg/source/testdata/plugin/dfs.go b/pkg/source/testdata/plugin/dfs.go
index 80dd05b0..5de6f779 100644
--- a/pkg/source/testdata/plugin/dfs.go
+++ b/pkg/source/testdata/plugin/dfs.go
@@ -22,6 +22,8 @@ import (
 	"io"
 	"io/ioutil"
 
+	"d7y.io/dragonfly/v2/pkg/util/rangeutils"
+
 	"d7y.io/dragonfly/v2/pkg/source"
 )
 
@@ -32,7 +34,7 @@ var _ source.ResourceClient = (*client)(nil)
 type client struct {
 }
 
-func (c *client) GetContentLength(ctx context.Context, url string, header source.RequestHeader) (int64, error) {
+func (c *client) GetContentLength(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (int64, error) {
 	return int64(len(data)), nil
 }
 
@@ -44,11 +46,11 @@ func (c *client) IsExpired(ctx context.Context, url string, header source.Reques
 	panic("implement me")
 }
 
-func (c *client) Download(ctx context.Context, url string, header source.RequestHeader) (io.ReadCloser, error) {
+func (c *client) Download(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, error) {
 	return ioutil.NopCloser(bytes.NewBufferString(data)), nil
 }
 
-func (c *client) DownloadWithResponseHeader(ctx context.Context, url string, header source.RequestHeader) (io.ReadCloser, source.ResponseHeader, error) {
+func (c *client) DownloadWithResponseHeader(ctx context.Context, url string, header source.RequestHeader, rang *rangeutils.Range) (io.ReadCloser, source.ResponseHeader, error) {
 	return ioutil.NopCloser(bytes.NewBufferString(data)), map[string]string{}, nil
 }