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

[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 introduces support for modern SMBv2/v3 protocols (via the smbj library) in the WindowsShareCopy plugin, while maintaining backward compatibility with legacy SMBv1 (via jcifs). It adds configuration options, path normalization, and corresponding unit tests. Feedback on the changes highlights a potential NullPointerException when converting a null password to a character array, as well as a performance bottleneck where new SMB connections and sessions are established per file in parallel instead of being reused across threads.

[gemini-code-assist]

Reusing the SMB connection, session, and share across all threads is highly recommended. In the current implementation, a new TCP connection is established and NTLM authentication is performed for every single file in parallel inside the executor threads. This introduces a massive performance bottleneck, especially when copying a large number of files, and can easily overwhelm the SMB server or lead to socket exhaustion.

Additionally, if config.netBiosPassword is null (e.g., for guest/anonymous access or during macro evaluation), calling config.netBiosPassword.toCharArray() will throw a NullPointerException. Guarding against this prevents runtime failures.

  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 == null ? new char[0] : 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 (Throwable t) {
              throw Throwables.propagate(t);
            }
            count++;
          }
        } finally {
          executorService.shutdownNow();
          executorTerminateLatch.await();
        }
      }
    }
  }