diff --git a/sdk/storage/azdatalake/CHANGELOG.md b/sdk/storage/azdatalake/CHANGELOG.md
index 1c764fbce113..a05617092be2 100644
--- a/sdk/storage/azdatalake/CHANGELOG.md
+++ b/sdk/storage/azdatalake/CHANGELOG.md
@@ -7,6 +7,7 @@
 ### Breaking Changes
 
 ### Bugs Fixed
+* Escape paths for NewDirectoryClient and NewFileClient in a file system. Fixes [#22281](https://github.com/Azure/azure-sdk-for-go/issues/22281).
 
 ### Other Changes
 
diff --git a/sdk/storage/azdatalake/assets.json b/sdk/storage/azdatalake/assets.json
index 626d6568abd4..a700b917992b 100644
--- a/sdk/storage/azdatalake/assets.json
+++ b/sdk/storage/azdatalake/assets.json
@@ -2,5 +2,5 @@
   "AssetsRepo": "Azure/azure-sdk-assets",
   "AssetsRepoPrefixPath": "go",
   "TagPrefix": "go/storage/azdatalake",
-  "Tag": "go/storage/azdatalake_3ae5e1441b"
+  "Tag": "go/storage/azdatalake_82ea48120e"
 }
diff --git a/sdk/storage/azdatalake/directory/client_test.go b/sdk/storage/azdatalake/directory/client_test.go
index 7214cdfd9726..6fb6b6e2eaa8 100644
--- a/sdk/storage/azdatalake/directory/client_test.go
+++ b/sdk/storage/azdatalake/directory/client_test.go
@@ -201,6 +201,35 @@ func (s *RecordedTestSuite) TestGetAndCreateFileClient() {
 	_require.NoError(err)
 }
 
+func (s *RecordedTestSuite) TestGetAndCreateFileClientWithSpecialNames() {
+	_require := require.New(s.T())
+	testName := s.T().Name()
+
+	filesystemName := testcommon.GenerateFileSystemName(testName)
+	fsClient, err := testcommon.GetFileSystemClient(filesystemName, s.T(), testcommon.TestAccountDatalake, nil)
+	_require.NoError(err)
+	defer testcommon.DeleteFileSystem(context.Background(), _require, fsClient)
+
+	_, err = fsClient.Create(context.Background(), nil)
+	_require.NoError(err)
+
+	dirClient := fsClient.NewDirectoryClient("#,%,?")
+	_require.NoError(err)
+
+	defer testcommon.DeleteDir(context.Background(), _require, dirClient)
+
+	resp, err := dirClient.Create(context.Background(), nil)
+	_require.NoError(err)
+	_require.NotNil(resp)
+
+	fileClient, err := dirClient.NewFileClient("?%#,")
+	_require.NoError(err)
+	_require.NotNil(fileClient)
+
+	_, err = fileClient.Create(context.Background(), nil)
+	_require.NoError(err)
+}
+
 func (s *RecordedTestSuite) TestCreateNewSubdirectoryClient() {
 	_require := require.New(s.T())
 	testName := s.T().Name()
@@ -265,6 +294,41 @@ func (s *RecordedTestSuite) TestCreateNewSubdirectoryClient() {
 	_require.True(datalakeerror.HasCode(err, datalakeerror.PathNotFound))
 }
 
+func (s *RecordedTestSuite) TestCreateNewSubdirectoryClientWithSpecialName() {
+	_require := require.New(s.T())
+	testName := s.T().Name()
+
+	accountName, _ := testcommon.GetGenericAccountInfo(testcommon.TestAccountDatalake)
+	_require.Greater(len(accountName), 0)
+
+	svcClient, err := testcommon.GetServiceClient(s.T(), testcommon.TestAccountDatalake, nil)
+	_require.NoError(err)
+
+	fsName := testcommon.GenerateFileSystemName(testName)
+	fsClient := svcClient.NewFileSystemClient(fsName)
+
+	_, err = fsClient.Create(context.Background(), nil)
+	_require.NoError(err)
+
+	defer testcommon.DeleteFileSystem(context.Background(), _require, fsClient)
+
+	dirClient := fsClient.NewDirectoryClient("?#%")
+
+	_, err = dirClient.Create(context.Background(), nil)
+	_require.NoError(err)
+
+	subdirClient, err := dirClient.NewSubdirectoryClient(",%,#,?")
+	_require.NoError(err)
+
+	perm := "r-xr-x---"
+
+	_, err = subdirClient.Create(context.Background(), &directory.CreateOptions{
+		Permissions: &perm,
+		CPKInfo:     &testcommon.TestCPKByValue,
+	})
+	_require.NoError(err)
+}
+
 func (s *RecordedTestSuite) TestCreateDirWithNilAccessConditions() {
 	_require := require.New(s.T())
 	testName := s.T().Name()
diff --git a/sdk/storage/azdatalake/filesystem/client.go b/sdk/storage/azdatalake/filesystem/client.go
index b69f14ba29eb..4fc2e7878bb6 100644
--- a/sdk/storage/azdatalake/filesystem/client.go
+++ b/sdk/storage/azdatalake/filesystem/client.go
@@ -202,7 +202,7 @@ func (fs *Client) BlobURL() string {
 // The new directory.Client uses the same request policy pipeline as the Client.
 func (fs *Client) NewDirectoryClient(directoryPath string) *directory.Client {
 	directoryPath = strings.ReplaceAll(directoryPath, "\\", "/")
-	dirURL := runtime.JoinPaths(fs.generatedFSClientWithDFS().Endpoint(), directoryPath)
+	dirURL := runtime.JoinPaths(fs.generatedFSClientWithDFS().Endpoint(), shared.EscapeSplitPaths(directoryPath))
 	blobURL, dirURL := shared.GetURLs(dirURL)
 	return (*directory.Client)(base.NewPathClient(dirURL, blobURL, fs.containerClient().NewBlockBlobClient(directoryPath), fs.generatedFSClientWithDFS().InternalClient().WithClientName(exported.ModuleName), fs.sharedKey(), fs.identityCredential(), fs.getClientOptions()))
 }
@@ -211,7 +211,7 @@ func (fs *Client) NewDirectoryClient(directoryPath string) *directory.Client {
 // The new file.Client uses the same request policy pipeline as the Client.
 func (fs *Client) NewFileClient(filePath string) *file.Client {
 	filePath = strings.ReplaceAll(filePath, "\\", "/")
-	fileURL := runtime.JoinPaths(fs.generatedFSClientWithDFS().Endpoint(), filePath)
+	fileURL := runtime.JoinPaths(fs.generatedFSClientWithDFS().Endpoint(), shared.EscapeSplitPaths(filePath))
 	blobURL, fileURL := shared.GetURLs(fileURL)
 	return (*file.Client)(base.NewPathClient(fileURL, blobURL, fs.containerClient().NewBlockBlobClient(filePath), fs.generatedFSClientWithDFS().InternalClient().WithClientName(exported.ModuleName), fs.sharedKey(), fs.identityCredential(), fs.getClientOptions()))
 }
diff --git a/sdk/storage/azdatalake/filesystem/client_test.go b/sdk/storage/azdatalake/filesystem/client_test.go
index c5bad8f0628a..a1f42713d8cd 100644
--- a/sdk/storage/azdatalake/filesystem/client_test.go
+++ b/sdk/storage/azdatalake/filesystem/client_test.go
@@ -8,6 +8,7 @@ package filesystem_test
 
 import (
 	"context"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azdatalake/file"
 	"strconv"
 	"strings"
 	"testing"
@@ -1318,6 +1319,90 @@ func (s *UnrecordedTestSuite) TestSASFileSystemClient() {
 	_require.NoError(err)
 }
 
+func (s *RecordedTestSuite) TestCreateFileSystemDirectoryClientWithSpecialDirName() {
+	_require := require.New(s.T())
+	testName := s.T().Name()
+
+	filesystemName := testcommon.GenerateFileSystemName(testName)
+	fsClient, err := testcommon.GetFileSystemClient(filesystemName, s.T(), testcommon.TestAccountDatalake, nil)
+	_require.NoError(err)
+	defer testcommon.DeleteFileSystem(context.Background(), _require, fsClient)
+
+	_, err = fsClient.Create(context.Background(), nil)
+	_require.NoError(err)
+
+	dirClient := fsClient.NewDirectoryClient("#,%,?/%,#")
+	_require.NoError(err)
+
+	response, err := dirClient.Create(context.Background(), nil)
+	_require.NoError(err)
+	_require.NotNil(response)
+
+	// Perform an operation on dfs endpoint
+	owner := "4cf4e284-f6a8-4540-b53e-c3469af032dc"
+	_, err = dirClient.SetAccessControl(context.Background(), &file.SetAccessControlOptions{Owner: &owner})
+	_require.NoError(err)
+
+	// Perform an operation on blob endpoint
+	_, err = dirClient.SetMetadata(context.Background(), testcommon.BasicMetadata, nil)
+	_require.NoError(err)
+}
+
+func (s *RecordedTestSuite) TestCreateFileSystemFileClientWithSpecialFileName() {
+	_require := require.New(s.T())
+	testName := s.T().Name()
+
+	filesystemName := testcommon.GenerateFileSystemName(testName)
+	fsClient, err := testcommon.GetFileSystemClient(filesystemName, s.T(), testcommon.TestAccountDatalake, nil)
+	_require.NoError(err)
+	defer testcommon.DeleteFileSystem(context.Background(), _require, fsClient)
+
+	_, err = fsClient.Create(context.Background(), nil)
+	_require.NoError(err)
+
+	fileClient := fsClient.NewFileClient("#,%,?/#")
+	_require.NoError(err)
+
+	response, err := fileClient.Create(context.Background(), nil)
+	_require.NoError(err)
+	_require.NotNil(response)
+
+	// Perform an operation on dfs endpoint
+	owner := "4cf4e284-f6a8-4540-b53e-c3469af032dc"
+	_, err = fileClient.SetAccessControl(context.Background(), &file.SetAccessControlOptions{Owner: &owner})
+	_require.NoError(err)
+
+	// Perform an operation on blob endpoint
+	_, err = fileClient.SetMetadata(context.Background(), testcommon.BasicMetadata, nil)
+	_require.NoError(err)
+}
+
+func (s *RecordedTestSuite) TestCreateFileClientFromDirectoryClientWithSpecialFileName() {
+	_require := require.New(s.T())
+	testName := s.T().Name()
+
+	filesystemName := testcommon.GenerateFileSystemName(testName)
+	fsClient, err := testcommon.GetFileSystemClient(filesystemName, s.T(), testcommon.TestAccountDatalake, nil)
+	_require.NoError(err)
+	defer testcommon.DeleteFileSystem(context.Background(), _require, fsClient)
+
+	_, err = fsClient.Create(context.Background(), nil)
+	_require.NoError(err)
+
+	dirClient := fsClient.NewDirectoryClient("#,%,?/%,#")
+	_require.NoError(err)
+
+	fileClient, err := dirClient.NewFileClient("%,?/##")
+	_require.NoError(err)
+
+	response, err := fileClient.Create(context.Background(), nil)
+	_require.NoError(err)
+	_require.NotNil(response)
+
+	_, err = fileClient.SetMetadata(context.Background(), testcommon.BasicMetadata, nil)
+	_require.NoError(err)
+}
+
 func (s *RecordedTestSuite) TestFilesystemListPathsWithRecursive() {
 	_require := require.New(s.T())
 	testName := s.T().Name()
diff --git a/sdk/storage/azdatalake/internal/shared/shared.go b/sdk/storage/azdatalake/internal/shared/shared.go
index d26008a79adb..9c964ee7ae56 100644
--- a/sdk/storage/azdatalake/internal/shared/shared.go
+++ b/sdk/storage/azdatalake/internal/shared/shared.go
@@ -240,3 +240,14 @@ func IsIPEndpointStyle(host string) bool {
 	}
 	return net.ParseIP(host) != nil
 }
+
+// EscapeSplitPaths is utility function to escape the individual strings by eliminating "/" in the path
+func EscapeSplitPaths(filePath string) string {
+	names := strings.Split(filePath, "/")
+	path := make([]string, len(names))
+	for i, name := range names {
+		path[i] = url.PathEscape(name)
+	}
+	escapedPathUrl := strings.Join(path, "/")
+	return escapedPathUrl
+}