Skip to content

[PLUGIN-1960] Add SMBv2/v3 support to WindowsShareCopy with legacy SMBv1 fallback #1997

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
vishwasvaidya-cloudsufi wants to merge 2 commits into
base: develop
Choose a base branch
from
+211 −19
Open

[PLUGIN-1960] Add SMBv2/v3 support to WindowsShareCopy with legacy SMBv1 fallback #1997

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d173f32
Migrate WindowsShareCopy from jcifs to smbj for SMBv2/v3 support
vishwasvaidya-cloudsufi Jul 20, 2026
5b8b73d
Add SMBv2/v3 support to WindowsShareCopy with backward compatibility
vishwasvaidya-cloudsufi Jul 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 20 additions & 4 deletions core-plugins/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,11 +111,27 @@
<groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId>
</dependency>
<!-- Windows share copy action -->
<!-- Windows share copy action (SMBv2/v3 support) -->
<dependency>
<groupId>jcifs</groupId>
<artifactId>jcifs</artifactId>
<version>1.3.17</version>
<groupId>com.hierynomus</groupId>
<artifactId>smbj</artifactId>
<version>0.11.5</version>
</dependency>
<!-- smbj transitive dependencies required for CDAP classloader -->
<dependency>
<groupId>net.engio</groupId>
<artifactId>mbassador</artifactId>
<version>1.3.0</version>

@vikasrathee-cs vikasrathee-cs Jul 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

make sure these libs doesn't have any vulnerabilities

</dependency>
<dependency>
<groupId>com.hierynomus</groupId>
<artifactId>asn-one</artifactId>
<version>0.6.0</version>
</dependency>
<dependency>
<groupId>org.bouncycastle</groupId>
<artifactId>bcprov-jdk15to18</artifactId>

@vikasrathee-cs vikasrathee-cs Jul 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we are having jdk 8, will there be any concern while using this?

@vishwasvaidya-cloudsufi vishwasvaidya-cloudsufi Jul 24, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Bouncy Castle uses a specific naming convention where the suffix indicates the supported Java versions. jdk15to18 explicitly means the artifact is compiled for and fully supports Java 1.5 through Java 1.8. Since we are on JDK 8, this is 100% compatible and exactly the intended artifact for our environment.

<version>1.68</version>
</dependency>
<!-- End for windows share copy action -->
<!-- EMail action -->
Expand Down
159 changes: 95 additions & 64 deletions core-plugins/src/main/java/io/cdap/plugin/batch/action/WindowsShareCopy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,16 @@
import com.google.common.base.Throwables;
import com.google.common.io.ByteStreams;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.hierynomus.msdtyp.AccessMask;
import com.hierynomus.msfscc.fileinformation.FileIdBothDirectoryInformation;
import com.hierynomus.mssmb2.SMB2CreateDisposition;
import com.hierynomus.mssmb2.SMB2ShareAccess;
import com.hierynomus.smbj.SMBClient;
import com.hierynomus.smbj.auth.AuthenticationContext;
import com.hierynomus.smbj.connection.Connection;
import com.hierynomus.smbj.session.Session;
import com.hierynomus.smbj.share.DiskShare;
import com.hierynomus.smbj.share.File;
import io.cdap.cdap.api.annotation.Description;
import io.cdap.cdap.api.annotation.Macro;
import io.cdap.cdap.api.annotation.Name;
Expand All @@ -30,8 +40,6 @@
import io.cdap.cdap.etl.api.StageConfigurer;
import io.cdap.cdap.etl.api.action.Action;
import io.cdap.cdap.etl.api.action.ActionContext;
import jcifs.smb.NtlmPasswordAuthentication;
import jcifs.smb.SmbFile;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
Expand All @@ -41,6 +49,9 @@
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.EnumSet;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.CountDownLatch;
Expand All @@ -54,6 +65,7 @@

/**
* {@link WindowsShareCopy} is an {@link Action} that will copy the data from a Windows share into an HDFS directory.
* Supports SMBv2 and SMBv3 via the smbj library.
*/
@Plugin(type = Action.PLUGIN_TYPE)
@Name("WindowsShareCopy")
Expand Down Expand Up @@ -81,100 +93,119 @@ public void run(ActionContext context) throws Exception {
config.numThreads;
config.bufferSize = (config.bufferSize == null || config.bufferSize < MIN_BUFFER_SIZE) ? MIN_BUFFER_SIZE :
config.bufferSize;
StringBuilder sb = new StringBuilder("smb://");
sb.append(config.netBiosHostname);
sb.append("/");
sb.append(config.netBiosSharename);
sb.append("/");
if (config.sourceDirectory.startsWith("/")) {
if (config.sourceDirectory.length() > 1) {
sb.append(config.sourceDirectory.substring(1));
}
} else {
sb.append(config.sourceDirectory);
// Normalize source directory to be compatible with smbj (remove leading slashes, use backslashes)
String normalizedSourcePath = config.sourceDirectory;
if (normalizedSourcePath.startsWith("/") || normalizedSourcePath.startsWith("\\")) {
normalizedSourcePath = normalizedSourcePath.substring(1);
}
final String smbDirectory = sb.toString();

// Register the SMB File handler.
jcifs.Config.registerSmbURLHandler();
// Set Jcifs Log level to log debug information also
jcifs.Config.setProperty("jcifs.util.loglevel", "4");

// Authentication with NTLM and read the directory from the Windows Share.
final NtlmPasswordAuthentication auth = new NtlmPasswordAuthentication(config.netBiosDomainName,
config.netBiosUsername,
config.netBiosPassword);
SmbFile dir = new SmbFile(smbDirectory, auth);
normalizedSourcePath = normalizedSourcePath.replace('/', '\\');

// Determines the buffer size to be used during writing.
Configuration conf = new Configuration();
conf.setLong("io.file.buffer.size", config.bufferSize);
conf.setLong("file.stream-buffer-size", config.bufferSize);

// Create the HDFS directory if it doesn't exist.
final Path hdfsDir = new Path(config.destinationDirectory);
final FileSystem hdfs = hdfsDir.getFileSystem(conf);
if (!hdfs.exists(hdfsDir)) {
hdfs.mkdirs(hdfsDir);
}

String[] files = dir.list();
// Copies the files in a multithreaded way
CountDownLatch executorTerminateLatch = new CountDownLatch(1);
ExecutorService executorService = createExecutor(config.numThreads, executorTerminateLatch);
CompletionService<String> completionService = new ExecutorCompletionService<>(executorService);

try {
for (final String file : files) {
completionService.submit(new Callable<String>() {
@Override
public String call() throws Exception {
try {
if (smbDirectory.endsWith("/")) {
return copyFileToHDFS(hdfs, smbDirectory + file, hdfsDir, auth);
} else {
return copyFileToHDFS(hdfs, smbDirectory + "/" + file, hdfsDir, auth);
}
} catch (Exception e) {
LOG.warn("Exception while copying the file {}", file, e);
return null;
SMBClient client = new SMBClient();
try (Connection connection = client.connect(config.netBiosHostname)) {
AuthenticationContext authContext = new AuthenticationContext(
config.netBiosUsername,
config.netBiosPassword.toCharArray(),
config.netBiosDomainName == null ? "" : config.netBiosDomainName
);

Session session = connection.authenticate(authContext);

try (DiskShare share = (DiskShare) session.connectShare(config.netBiosSharename)) {

@gemini-code-assist gemini-code-assist Bot Jul 20, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

The SMBClient and Session instances are not closed, which can lead to thread leaks (since SMBClient manages its own executor service) and orphaned sessions on the SMB server. Both classes implement AutoCloseable and should be managed using try-with-resources to ensure they are properly closed.

    try (SMBClient client = new SMBClient();
         Connection connection = client.connect(config.netBiosHostname)) {
      AuthenticationContext authContext = new AuthenticationContext(
              config.netBiosUsername,
              config.netBiosPassword.toCharArray(),
              config.netBiosDomainName == null ? "" : config.netBiosDomainName
      );

      try (Session session = connection.authenticate(authContext);
           DiskShare share = (DiskShare) session.connectShare(config.netBiosSharename)) {


List<String> filesToCopy = new ArrayList<>();

// Determine if path is a folder or single file
if (share.folderExists(normalizedSourcePath)) {
for (FileIdBothDirectoryInformation f : share.list(normalizedSourcePath)) {
String fileName = f.getFileName();
if (fileName.equals(".") || fileName.equals("..")) {
continue;
}
String fullPath = normalizedSourcePath.isEmpty() ? fileName : normalizedSourcePath + "\\" + fileName;
filesToCopy.add(fullPath);
}
});
}
} else if (share.fileExists(normalizedSourcePath)) {
filesToCopy.add(normalizedSourcePath);
} else {
throw new IllegalArgumentException(
String.format("The source path '%s' does not exist on share '%s'", normalizedSourcePath,
config.netBiosSharename)
);
}
// Copies the files in a multithreaded way
CountDownLatch executorTerminateLatch = new CountDownLatch(1);
ExecutorService executorService = createExecutor(config.numThreads, executorTerminateLatch);
CompletionService<String> completionService = new ExecutorCompletionService<>(executorService);

int count = 0;
while (count < files.length) {
try {
Future<String> fileWritten = completionService.take();
String fileName = fileWritten.get();
if (fileName != null) {
LOG.debug("{} is copied", fileName);
for (final String filePath : filesToCopy) {
completionService.submit(new Callable<String>() {
@Override
public String call() throws Exception {
try {
return copyFileToHDFS(hdfs, share, filePath, hdfsDir);
} catch (Exception e) {
LOG.warn("Exception while copying the file {}", filePath, e);
return null;
}
}
});
}

int count = 0;
while (count < filesToCopy.size()) {
try {
Future<String> fileWritten = completionService.take();
String fileName = fileWritten.get();
if (fileName != null) {
LOG.debug("{} is copied", fileName);
}
} catch (Throwable t) {
throw Throwables.propagate(t);
}
count++;
}
} catch (Throwable t) {
throw Throwables.propagate(t);
} finally {
executorService.shutdownNow();
executorTerminateLatch.await();
}
count++;
}
} finally {
executorService.shutdownNow();
executorTerminateLatch.await();
}
}

private String copyFileToHDFS(FileSystem hdfs, String smbSourceFile, Path dest, NtlmPasswordAuthentication auth)
private String copyFileToHDFS(FileSystem hdfs, DiskShare share, String smbSourceFile, Path dest)
throws IOException {
SmbFile smbFile = new SmbFile(smbSourceFile, auth);
String name = smbFile.getName();
int lastSlashIndex = smbSourceFile.lastIndexOf('\\');
String name = lastSlashIndex >= 0 ? smbSourceFile.substring(lastSlashIndex + 1) : smbSourceFile;
Path destFile = new Path(dest, name);
LOG.debug("Thread {} is copying source file {}, dest file {}", Thread.currentThread().getName(),
smbFile.getName(), destFile.getName());
name, destFile.getName());
// If file already exists, then we skip over.
if (hdfs.exists(destFile) && !config.overwrite) {
LOG.info("File {} already exists on HDFS, Skipping", destFile.getName());
return null;
}
LOG.info("Copying file {} to {}", smbSourceFile, destFile.toString());
try (InputStream in = smbFile.getInputStream();
try (File smbFile = share.openFile(
smbSourceFile,
EnumSet.of(AccessMask.GENERIC_READ),
null,
SMB2ShareAccess.ALL,
SMB2CreateDisposition.FILE_OPEN,
null);
InputStream in = smbFile.getInputStream();
BufferedOutputStream out = new BufferedOutputStream(hdfs.create(destFile), config.bufferSize)) {
ByteStreams.copy(in, out);
} catch (IOException e) {
Expand Down
Loading