diff --git a/core-plugins/pom.xml b/core-plugins/pom.xml
index 8bcee2fd8..74e7d5c49 100644
--- a/core-plugins/pom.xml
+++ b/core-plugins/pom.xml
@@ -111,13 +111,33 @@
commons-lang
commons-lang
-
+
jcifs
jcifs
1.3.17
-
+
+
+ com.hierynomus
+ smbj
+ 0.11.5
+
+
+ net.engio
+ mbassador
+ 1.3.0
+
+
+ com.hierynomus
+ asn-one
+ 0.6.0
+
+
+ org.bouncycastle
+ bcprov-jdk15to18
+ 1.84
+
dumbster
diff --git a/core-plugins/src/main/java/io/cdap/plugin/batch/action/WindowsShareCopy.java b/core-plugins/src/main/java/io/cdap/plugin/batch/action/WindowsShareCopy.java
index a683c65a2..3698021d6 100644
--- a/core-plugins/src/main/java/io/cdap/plugin/batch/action/WindowsShareCopy.java
+++ b/core-plugins/src/main/java/io/cdap/plugin/batch/action/WindowsShareCopy.java
@@ -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;
@@ -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;
@@ -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")
@@ -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("/");
@@ -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);
@@ -128,15 +156,10 @@ public void run(ActionContext context) throws Exception {
completionService.submit(new Callable() {
@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);
}
}
});
@@ -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());
@@ -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 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 completionService = new ExecutorCompletionService<>(executorService);
+
+ try {
+ for (final String filePath : filesToCopy) {
+ completionService.submit(new Callable() {
+ @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 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();
+ }
+ }
+ }
+
+ /**
+ * 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;
}
@@ -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;
@@ -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) {
diff --git a/core-plugins/src/test/java/io/cdap/plugin/batch/ETLBatchTestBase.java b/core-plugins/src/test/java/io/cdap/plugin/batch/ETLBatchTestBase.java
index 8c393c377..378f73810 100644
--- a/core-plugins/src/test/java/io/cdap/plugin/batch/ETLBatchTestBase.java
+++ b/core-plugins/src/test/java/io/cdap/plugin/batch/ETLBatchTestBase.java
@@ -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;
@@ -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
diff --git a/core-plugins/src/test/java/io/cdap/plugin/batch/action/WindowsShareCopyConfigTest.java b/core-plugins/src/test/java/io/cdap/plugin/batch/action/WindowsShareCopyConfigTest.java
new file mode 100644
index 000000000..a709c5d96
--- /dev/null
+++ b/core-plugins/src/test/java/io/cdap/plugin/batch/action/WindowsShareCopyConfigTest.java
@@ -0,0 +1,111 @@
+/*
+ * Copyright © 2026 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.plugin.batch.action;
+
+import io.cdap.cdap.etl.mock.validation.MockFailureCollector;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Unit tests for {@link WindowsShareCopy.WindowsShareCopyConfig}
+ */
+public class WindowsShareCopyConfigTest {
+
+ private WindowsShareCopy.WindowsShareCopyConfig createConfig(String sourceDirectory, String smbVersion) {
+ return new WindowsShareCopy(null).new WindowsShareCopyConfig(
+ "testDomain",
+ "testHostname",
+ "testUser",
+ "testPassword",
+ "testShare",
+ sourceDirectory,
+ "/testDest",
+ 4096,
+ 1,
+ "true",
+ smbVersion
+ );
+ }
+
+ @Test
+ public void testPathNormalization() {
+ // 1. Test multiple leading forward slashes
+ WindowsShareCopy.WindowsShareCopyConfig config1 = createConfig("///folder/subfolder", "SMBv2/v3");
+ Assert.assertEquals("folder\\subfolder", config1.getSourceDirectory());
+
+ // 2. Test multiple leading backward slashes
+ WindowsShareCopy.WindowsShareCopyConfig config2 = createConfig("\\\\\\folder\\subfolder", "SMBv2/v3");
+ Assert.assertEquals("folder\\subfolder", config2.getSourceDirectory());
+
+ // 3. Test mixed/no leading slashes but internal forward slashes
+ WindowsShareCopy.WindowsShareCopyConfig config3 = createConfig("folder/subfolder/file.txt", "SMBv2/v3");
+ Assert.assertEquals("folder\\subfolder\\file.txt", config3.getSourceDirectory());
+
+ // 4. Test empty string
+ WindowsShareCopy.WindowsShareCopyConfig config4 = createConfig("", "SMBv2/v3");
+ Assert.assertEquals("", config4.getSourceDirectory());
+
+ // 5. Test null
+ WindowsShareCopy.WindowsShareCopyConfig config5 = createConfig(null, "SMBv2/v3");
+ Assert.assertNull(config5.getSourceDirectory());
+ }
+
+ @Test
+ public void testSMBVersionFallback() {
+ // 1. Test null falls back to true (SMBv1) for backward compatibility
+ WindowsShareCopy.WindowsShareCopyConfig nullConfig = createConfig("/dir", null);
+ Assert.assertTrue(nullConfig.isSMBv1());
+
+ // 2. Test explicit SMBv1
+ WindowsShareCopy.WindowsShareCopyConfig v1Config = createConfig("/dir", "SMBv1");
+ Assert.assertTrue(v1Config.isSMBv1());
+
+ // 2b. Test explicit SMBv1 (case insensitive)
+ WindowsShareCopy.WindowsShareCopyConfig v1ConfigLower = createConfig("/dir", "smbv1");
+ Assert.assertTrue(v1ConfigLower.isSMBv1());
+
+ // 3. Test explicit SMBv2/v3
+ WindowsShareCopy.WindowsShareCopyConfig v2Config = createConfig("/dir", "SMBv2/v3");
+ Assert.assertFalse(v2Config.isSMBv1());
+ }
+
+ @Test
+ public void testValidConfiguration() {
+ WindowsShareCopy.WindowsShareCopyConfig config = createConfig("/dir", "SMBv2/v3");
+ MockFailureCollector collector = new MockFailureCollector();
+ config.validate(collector);
+
+ Assert.assertEquals(0, collector.getValidationFailures().size());
+ }
+
+ @Test
+ public void testInvalidConfiguration() {
+ // Create a config with null for all required fields
+ WindowsShareCopy.WindowsShareCopyConfig invalidConfig =
+ new WindowsShareCopy(null).new WindowsShareCopyConfig(
+ null, null, null, null,
+ null, null, null, 4096,
+ 1, "true", "SMBv2/v3"
+ );
+
+ MockFailureCollector collector = new MockFailureCollector();
+ invalidConfig.validate(collector);
+
+ // Should trigger 5 validation failures (hostname, username, password, sharename, sourcedir, destdir)
+ Assert.assertEquals(6, collector.getValidationFailures().size());
+ }
+}
diff --git a/core-plugins/widgets/WindowsShareCopy-action.json b/core-plugins/widgets/WindowsShareCopy-action.json
index 861297bf4..c47c60959 100644
--- a/core-plugins/widgets/WindowsShareCopy-action.json
+++ b/core-plugins/widgets/WindowsShareCopy-action.json
@@ -61,6 +61,18 @@
{
"label": "Plugin Properties",
"properties": [
+ {
+ "widget-type": "select",
+ "label": "SMB Protocol Version",
+ "name": "smbVersion",
+ "widget-attributes": {
+ "values": [
+ "SMBv2/v3",
+ "SMBv1"
+ ],
+ "default": "SMBv2/v3"
+ }
+ },
{
"widget-type": "textbox",
"label": "Number of parallel tasks",