Skip to content

[🍒] Add SMBv2/v3 support to WindowsShareCopy with legacy SMBv1 fallback for release/2.13 #2000

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 1 commit into
base: release/2.13
Choose a base branch
from
Open

[🍒] Add SMBv2/v3 support to WindowsShareCopy with legacy SMBv1 fallback for release/2.13 #2000

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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: 22 additions & 2 deletions core-plugins/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,13 +111,33 @@
<groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId>
</dependency>
<!-- Windows share copy action -->
<!-- Legacy SMBv1 Support -->
<dependency>
<groupId>jcifs</groupId>
<artifactId>jcifs</artifactId>
<version>1.3.17</version>
</dependency>
<!-- End for windows share copy action -->
<!-- Modern SMBv2/v3 Support -->
<dependency>
<groupId>com.hierynomus</groupId>
<artifactId>smbj</artifactId>
<version>0.11.5</version>
</dependency>
<dependency>
<groupId>net.engio</groupId>
<artifactId>mbassador</artifactId>
<version>1.3.0</version>
</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>
<version>1.84</version>
</dependency>
<!-- EMail action -->
<dependency>
<groupId>dumbster</groupId>
Expand Down
219 changes: 191 additions & 28 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 @@ -41,6 +51,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 +67,7 @@

/**
* {@link WindowsShareCopy} is an {@link Action} that will copy the data from a Windows share into an HDFS directory.
* Supports SMBv1 (via jcifs) and SMBv2/SMBv3 (via smbj) based on user configuration.
*/
@Plugin(type = Action.PLUGIN_TYPE)
@Name("WindowsShareCopy")
Expand Down Expand Up @@ -81,6 +95,32 @@ public void run(ActionContext context) throws Exception {
config.numThreads;
config.bufferSize = (config.bufferSize == null || config.bufferSize < MIN_BUFFER_SIZE) ? MIN_BUFFER_SIZE :
config.bufferSize;

// 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);
}

if (config.isSMBv1()) {
LOG.info("Executing copy using legacy SMBv1 protocol (jcifs)");
executeSMBv1(hdfsDir, hdfs);
} else {
LOG.info("Executing copy using modern SMBv2/v3 protocol (smbj)");
executeSMBv2(hdfsDir, hdfs);
}
}

/**
* Executes the copy operation using the legacy jcifs (SMBv1) library.
*/
private void executeSMBv1(final Path hdfsDir, final FileSystem hdfs) throws Exception {
StringBuilder sb = new StringBuilder("smb://");
sb.append(config.netBiosHostname);
sb.append("/");
Expand All @@ -102,21 +142,9 @@ public void run(ActionContext context) throws Exception {

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

// 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);
Expand All @@ -128,15 +156,10 @@ public void run(ActionContext context) throws Exception {
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;
if (smbDirectory.endsWith("/")) {
return copyFileToHDFS(hdfs, smbDirectory + file, hdfsDir, auth);
} else {
return copyFileToHDFS(hdfs, smbDirectory + "/" + file, hdfsDir, auth);
}
}
});
Expand All @@ -162,12 +185,12 @@ public String call() throws Exception {
}

private String copyFileToHDFS(FileSystem hdfs, String smbSourceFile, Path dest, NtlmPasswordAuthentication auth)
throws IOException {
throws IOException {
SmbFile smbFile = new SmbFile(smbSourceFile, auth);
String name = smbFile.getName();
Path destFile = new Path(dest, name);
LOG.debug("Thread {} is copying source file {}, dest file {}", Thread.currentThread().getName(),
smbFile.getName(), destFile.getName());
smbFile.getName(), 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());
Expand All @@ -177,8 +200,115 @@ private String copyFileToHDFS(FileSystem hdfs, String smbSourceFile, Path dest,
try (InputStream in = smbFile.getInputStream();
BufferedOutputStream out = new BufferedOutputStream(hdfs.create(destFile), config.bufferSize)) {
ByteStreams.copy(in, out);
} catch (IOException e) {
LOG.warn("Exception in copying the file {}", name, e);
}
return name;
}

/**
* Executes the copy operation using the modern smbj library.
*/
private void executeSMBv2(final Path hdfsDir, final FileSystem hdfs) throws Exception {
String normalizedSourcePath = config.getSourceDirectory();

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

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

try (Connection listConnection = client.connect(config.netBiosHostname);
Session listSession = listConnection.authenticate(authContext);
DiskShare listShare = (DiskShare) listSession.connectShare(config.netBiosSharename)) {

if (listShare.folderExists(normalizedSourcePath)) {
for (FileIdBothDirectoryInformation f : listShare.list(normalizedSourcePath)) {
String fileName = f.getFileName();
if (fileName.equals(".") || fileName.equals("..")) {
continue;
}
String fullPath = normalizedSourcePath.isEmpty() ? fileName : normalizedSourcePath + "\\" + fileName;
filesToCopy.add(fullPath);
}
} else if (listShare.fileExists(normalizedSourcePath)) {
filesToCopy.add(normalizedSourcePath);
} else {
throw new IllegalArgumentException(
String.format("The source path '%s' does not exist on share '%s'",
normalizedSourcePath, config.netBiosSharename)
);
}
}

CountDownLatch executorTerminateLatch = new CountDownLatch(1);
ExecutorService executorService = createExecutor(config.numThreads, executorTerminateLatch);
CompletionService<String> completionService = new ExecutorCompletionService<>(executorService);

try {
for (final String filePath : filesToCopy) {
completionService.submit(new Callable<String>() {
@Override
public String call() throws Exception {
try (Connection threadConnection = client.connect(config.netBiosHostname);
Session threadSession = threadConnection.authenticate(authContext);
DiskShare threadShare = (DiskShare) threadSession.connectShare(config.netBiosSharename)) {

return copyFileToHDFS_SMBv2(hdfs, threadShare, filePath, hdfsDir);
}
}
});
}

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++;
}
} finally {
executorService.shutdownNow();
executorTerminateLatch.await();
}
}
}
Comment on lines +210 to +282

@gemini-code-assist gemini-code-assist Bot Jul 29, 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

In the current implementation of executeSMBv2, a new SMB connection, session, and share are established inside the Callable for every single file being copied. This introduces massive performance overhead due to repeated TCP handshakes, SMB negotiation, and NTLM authentication for each file. It can also lead to socket exhaustion or trigger connection limits/security blocks on the target Windows/Active Directory server.

Since smbj's DiskShare, Session, and Connection are thread-safe and designed to multiplex concurrent requests, you should establish the connection, session, and share once at the method level, use it to list the files, and then share the same DiskShare instance across all concurrent copy tasks. This will drastically improve performance and resource utilization.

Additionally, this suggestion replaces the deprecated Throwables.propagate(t) with standard Java exception handling.

private void executeSMBv2(final Path hdfsDir, final FileSystem hdfs) throws Exception {
    String normalizedSourcePath = config.getSourceDirectory();

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

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

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

        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)
          );
        }

        CountDownLatch executorTerminateLatch = new CountDownLatch(1);
        ExecutorService executorService = createExecutor(config.numThreads, executorTerminateLatch);
        CompletionService<String> completionService = new ExecutorCompletionService<>(executorService);

        try {
          for (final String filePath : filesToCopy) {
            completionService.submit(new Callable<String>() {
              @Override
              public String call() throws Exception {
                return copyFileToHDFS_SMBv2(hdfs, share, filePath, hdfsDir);
              }
            });
          }

          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 (ExecutionException e) {
              throw new RuntimeException(e.getCause());
            } catch (InterruptedException e) {
              Thread.currentThread().interrupt();
              throw new RuntimeException(e);
            }
            count++;
          }
        } finally {
          executorService.shutdownNow();
          executorTerminateLatch.await();
        }
      }
    }
  }


/**
* The new SMBv2/v3 file copy method
*/
private String copyFileToHDFS_SMBv2(FileSystem hdfs, DiskShare share, String smbSourceFile, Path dest)
throws IOException {
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(),
name, destFile.getName());

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 (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);
}
return name;
}
Expand Down Expand Up @@ -263,9 +393,16 @@ public class WindowsShareCopyConfig extends PluginConfig {
@Macro
private Integer bufferSize;

@Description("Specifies the SMB protocol version. Use 'SMBv1' for legacy systems or 'SMBv2/v3' for modern " +
"secure shares.")
@Nullable
@Macro
private final String smbVersion;

WindowsShareCopyConfig(String netBiosDomainName, String netBiosHostname, String netBiosUsername,
String netBiosPassword, String netBiosSharename, String sourceDirectory,
String destinationDirectory, Integer bufferSize, Integer numThreads, String overwrite) {
String destinationDirectory, Integer bufferSize, Integer numThreads, String overwrite
, String smbVersion) {

this.netBiosDomainName = netBiosDomainName;
this.netBiosHostname = netBiosHostname;
Expand All @@ -277,6 +414,32 @@ public class WindowsShareCopyConfig extends PluginConfig {
this.bufferSize = bufferSize;
this.numThreads = numThreads;
this.overwrite = !("false".equals(overwrite));
this.smbVersion = smbVersion;
}

/**
* Determines if the plugin should use legacy SMBv1.
* Defaults to true (SMBv1) if null, meaning old pipelines will work without modification.
*/
public boolean isSMBv1() {
return smbVersion == null || "SMBv1".equalsIgnoreCase(smbVersion);
}

/**
* Normalizes the source directory by removing all leading slashes
* and converting to Windows-native backslashes for SMBv2/v3.
*/
public String getSourceDirectory() {
if (Strings.isNullOrEmpty(sourceDirectory)) {
return sourceDirectory;
}
String normalized = sourceDirectory;

while (normalized.startsWith("/") || normalized.startsWith("\\")) {
normalized = normalized.substring(1);
}

return normalized.replace('/', '\\');
}

public void validate(FailureCollector collector) {
Expand Down
2 changes: 2 additions & 0 deletions core-plugins/src/test/java/io/cdap/plugin/batch/ETLBatchTestBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@
import io.cdap.plugin.alert.TMSAlertPublisher;
import io.cdap.plugin.batch.action.EmailAction;
import io.cdap.plugin.batch.action.SSHAction;
import io.cdap.plugin.batch.action.WindowsShareCopy;
import io.cdap.plugin.batch.aggregator.DedupAggregator;
import io.cdap.plugin.batch.aggregator.GroupByAggregator;
import io.cdap.plugin.batch.aggregator.function.AggregateFunction;
Expand Down Expand Up @@ -147,6 +148,7 @@ public static void setupTest() throws Exception {
TMSAlertPublisher.class,
ErrorCollector.class,
FileSink.class,
WindowsShareCopy.class,
// Spark needs this class to get exported to use the correct classloader
AggregateFunction.class);
// add format plugins
Expand Down
Loading