[shruthi713]

Summary

Launches the singleton TaskWorkerManager Netty proxy as a standalone Twill application from App Fabric when TaskWorkerManager.isEnabled() is true.

• TaskWorkerManagerServiceLauncher: Scheduled service that launches TaskWorkerManagerTwillApplication, manages instance lifecycles, sets withRecreateStrategy(), strips unmounted internal certificates, and passes ENDPOINTS_SERVICES = "task.worker" in cConfCopy so the proxy container discovers worker pods directly.
• TaskWorkerManagerTwillApplication: Standalone Twill application specifying the proxy runnable with 1 replica.
• TaskWorkerManagerTwillRunnable: Boots TaskWorkerManagerService with Guice injector in the proxy container.
• AppFabricProcessorServiceMain: Adds TaskWorkerManagerServiceLauncher to the scheduled services when TaskWorkerManager.isEnabled(cConf) holds.
• RBAC prerequisite (when enabling the feature flag): The CDAP Kubernetes Role (cdap-role) must grant get, list, and watch on endpoints so the proxy container's endpoints watcher (feat(k8s): watch V1Endpoints in KubeDiscoveryService #16226) can resolve individual task.worker pod IPs.
• Unit Tests: TaskWorkerManagerServiceLauncherTest (12 tests) verifying singleton controller adoption, recreate rollout strategy, CConfiguration endpoints discovery setup, the readiness probe (HTTP and HTTPS schemes), priority class setting, shutDown() cleanup, and the drain time limits; KubeTwillPreparerTest covers the new grace period.

Proxy restarts and planned stops

• Readiness probe. Without one, Kubernetes marked a new proxy pod ready as soon as its container started, several seconds before the JVM listened, so callers were refused during every restart. The launcher now passes task.worker.manager.probe.* to addProbes, the mechanism the task workers already use, and sets k8s.readiness.http.scheme to HTTPS when ssl.internal.enabled is true (HTTP when false; kubelet skips certificate verification on HTTPS probe checks). The default is a readiness probe on the proxy's own GET /ping (bind port, every 2s, failure threshold 3 so one slow answer under load doesn't stop routing). There is deliberately no liveness probe: restarting the proxy would cut every in-flight task. task.worker.manager.probe.enabled=false turns it off.
• Time to drain. The proxy now drains in-flight requests on a graceful stop (PR D), but two outer limits would cut it short at 30s: the JVM shutdown hook in MasterEnvironmentMain, and Kubernetes' default termination grace period.
  • The hook's wait is now master.environment.k8s.runnable.stop.timeout.secs. It defaults to 30, so every other runnable is unchanged.
  • ExtendedTwillPreparer gets withTerminationGracePeriod, which sets the pod's terminationGracePeriodSeconds on a Deployment. Unset keeps the cluster default.
  • The launcher derives both from task.worker.manager.drain.timeout.secs, for the proxy only: its stop wait is drain + 10s and its grace period drain + 15s (70s and 75s by default).

Part of the Task Worker Manager PR stack (merge order: A #16225 and B #16226 first, then C #16229 → D #16230 → E #16231 → F #16232):

• Prerequisite: PR feat(twill): add withRecreateStrategy to ExtendedTwillPreparer #16225 (Recreate Strategy)
• Prerequisite: PR feat(k8s): watch V1Endpoints in KubeDiscoveryService #16226 (Endpoints Discovery)
• Stacked on PR E: feat(rbac): route task worker traffic through the manager proxy #16231 (Routing & Admission Control)

[gemini-code-assist]

Code Review

This pull request introduces the TaskWorkerManagerServiceLauncher, TaskWorkerManagerTwillApplication, and TaskWorkerManagerTwillRunnable to launch and manage the task worker manager netty proxy. It also extends KubeTwillPreparer to support a 'Recreate' deployment strategy, ensuring at-most-one pod runs during updates. Feedback on the changes highlights several critical improvements: propagating service failures in the runnable instead of swallowing them, annotating the launcher with @singleton, declaring twillController as volatile for thread safety, properly restoring the interrupted status when handling InterruptedException, and using DirUtils.deleteDirectory to prevent leaking empty temporary directories.

[gemini-code-assist]

When the taskManagerService fails, Uninterruptibles.getUninterruptibly(future) throws an ExecutionException. Currently, this exception is caught, logged as a warning, and the run() method exits normally. This causes the Twill container to exit with a success status (code 0), preventing Kubernetes or Twill from detecting the failure and triggering restarts or alerts.

Please propagate the failure by throwing a RuntimeException wrapping the cause of the ExecutionException.

    try {
      Uninterruptibles.getUninterruptibly(future);
      LOG.debug("Task worker manager stopped");
    } catch (ExecutionException e) {
      LOG.error("Task worker manager stopped with exception", e);
      throw new RuntimeException("Task worker manager service failed", e.getCause());
    }

[shruthi713]

Fixed in 77eb49b. The ExecutionException cause is now logged and rethrown, so a crashed proxy fails the runnable instead of looking like a clean stop.

[gemini-code-assist]

The TaskWorkerManagerServiceLauncher is a scheduled service that manages a singleton Twill application. To prevent accidental multiple instantiations within the Guice injector (since it is currently resolved via JIT binding), it should be annotated with @Singleton.

Suggested change

	public class TaskWorkerManagerServiceLauncher extends AbstractScheduledService {
	@com.google.inject.Singleton
	public class TaskWorkerManagerServiceLauncher extends AbstractScheduledService {

[shruthi713]

Fixed in 77eb49b; the launcher is now @Singleton.

[gemini-code-assist]

The twillController field is updated on the scheduled executor thread inside the run() method, but it is read on the service management thread inside the shutDown() method. Without the volatile modifier, there is no guarantee of memory visibility across threads, which could lead to shutDown() seeing a stale or null value and failing to terminate the active controller.

Suggested change

	private TwillController twillController;
	private volatile TwillController twillController;

[shruthi713]

volatile isn't needed here. AbstractScheduledService runs runOneIteration() and shutDown() under the same ReentrantLock (checked against Guava 13.0.1, which CDAP ships), so the write in runOneIteration() happens-before the read in shutDown(). TaskWorkerServiceLauncher and SystemWorkerServiceLauncher rely on the same guarantee with a non-volatile field.

[gemini-code-assist]

In shutDown(), calling twillController.terminate().get(10, TimeUnit.SECONDS) can throw an InterruptedException. If an InterruptedException is caught by the generic catch (Exception e) block, the thread's interrupted status is cleared and lost.

It is a best practice to catch InterruptedException explicitly and restore the interrupted status by calling Thread.currentThread().interrupt().

Suggested change

	try {
	if (twillController != null) {
	twillController.terminate().get(10, TimeUnit.SECONDS);
	}
	} catch (Exception e) {
	LOG.warn("Failed to terminate TaskWorkerManagerServiceLauncher run", e);
	}
	try {
	if (twillController != null) {
	twillController.terminate().get(10, TimeUnit.SECONDS);
	}
	} catch (InterruptedException e) {
	LOG.warn("Interrupted while terminating TaskWorkerManagerServiceLauncher run", e);
	Thread.currentThread().interrupt();
	} catch (Exception e) {
	LOG.warn("Failed to terminate TaskWorkerManagerServiceLauncher run", e);
	}

[shruthi713]

Fixed in 77eb49b; InterruptedException now restores the interrupt flag.

[gemini-code-assist]

The deleteDir method uses DirUtils.deleteDirectoryContents(dir.toFile()), which recursively deletes all files and subdirectories inside the specified directory, but leaves the empty directory itself. Since runDir is a temporary directory created via Files.createTempDirectory, this will leak empty directories in the system's temporary folder over time.

Please use DirUtils.deleteDirectory(dir.toFile()) instead, which deletes both the contents and the directory itself.

Suggested change

	private void deleteDir(Path dir) {
	try {
	if (Files.isDirectory(dir)) {
	DirUtils.deleteDirectoryContents(dir.toFile());
	}
	} catch (IOException e) {
	LOG.warn("Failed to cleanup directory {}", dir, e);
	}
	}
	private void deleteDir(Path dir) {
	try {
	if (Files.isDirectory(dir)) {
	DirUtils.deleteDirectory(dir.toFile());
	}
	} catch (IOException e) {
	LOG.warn("Failed to cleanup directory {}", dir, e);
	}
	}

[shruthi713]

This doesn't leak. DirUtils.deleteDirectoryContents(File) delegates to deleteDirectoryContents(path, retain = false), which also deletes the root directory in postVisitDirectory. The sibling launchers use the same call.