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

[vishwasvaidya-cloudsufi]

[🍒]

🍒 cherrypick

PR : (#1997)

Description:

Kept the legacy jcifs (SMBv1) library to maintain backward compatibility. The plugin dynamically routes to either SMBv1 or SMBv2/v3, defaulting to the newer protocol for all new pipelines.

Key Changes:

Backward Compatibility / UI: Added an SMB Protocol Version dropdown to the UI. New pipelines default to SMBv2/v3. Crucially, if the configuration receives a null value (which happens when existing pipelines are upgraded), the plugin safely falls back to SMBv1 to prevent pipeline breakage.

Dependency Management: Retained jcifs and added smbj along with its explicitly required transitive dependencies (mbassador, asn-one, bcprov) to the pom.xml to ensure CDAP classloader compatibility.

Modern SMB Architecture: Implemented a secure, hierarchical connection lifecycle (Client -> Session -> Share) for the new smbj execution path.

Path Normalization: Added logic to convert UNIX-style forward slashes (/) to Windows-native backslashes () for the smbj share traversal.

Directory Traversal: Implemented explicit directory vs. file traversal logic for SMBv2/v3, including filtering for Windows system pointer directories (. and ..).

Safe File Operations: Applied AccessMask.GENERIC_READ during smbj file operations to prevent accidental file locking on the destination Windows server.

[gemini-code-assist]

Code Review

This pull request adds support for SMBv2/v3 protocols using the smbj library in the WindowsShareCopy plugin, while maintaining backward compatibility with SMBv1. It introduces configuration options for selecting the SMB version, implements the SMBv2/v3 copy logic, and adds corresponding unit tests. Feedback was provided to address a performance issue in the SMBv2 implementation, where establishing a new connection, session, and share for every file copy task inside the executor service introduces significant overhead; sharing a single DiskShare instance across concurrent tasks is recommended instead.

[gemini-code-assist]

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