Runtime Providers and Program Execution

/**
 * Runtime provider for Spark 1.x programs in CDAP
 * Extends SparkProgramRuntimeProvider with Spark 1.x specific compatibility
 */
@ProgramRuntimeProvider.SupportedProgramType(ProgramType.SPARK)
public class Spark1ProgramRuntimeProvider extends SparkProgramRuntimeProvider {
    /**
     * Creates a Spark 1.x runtime provider with SPARK1_2_10 compatibility
     */
    public Spark1ProgramRuntimeProvider();
}

/**
 * Abstract base class for Spark program runtime providers  
 * Handles class loading, runner creation, and program lifecycle management
 */
public abstract class SparkProgramRuntimeProvider implements ProgramRuntimeProvider {
    /**
     * Creates a program runner for the specified program type and execution mode
     * @param type The type of program (must be SPARK)
     * @param mode Execution mode (LOCAL or DISTRIBUTED)
     * @param injector Guice injector for dependency injection
     * @return ProgramRunner instance for executing the program
     * @throws IllegalArgumentException if program type is not SPARK
     */
    public ProgramRunner createProgramRunner(ProgramType type, Mode mode, Injector injector);
    
    /**
     * Checks if the specified program type is supported
     * @param programType The program type to check
     * @param cConf CDAP configuration  
     * @return true if program type is SPARK
     */
    public boolean isSupported(ProgramType programType, CConfiguration cConf);
    
    /**
     * Gets the Spark compatibility version for this provider
     * @return SparkCompat enum value (SPARK1_2_10, SPARK2_2_11, etc.)
     */
    protected SparkCompat getSparkCompat();
}

/**
 * Controller interface for managing program execution lifecycle
 */
public interface ProgramController {
    /**
     * Sends a command to the running program
     * @param name Command name
     * @param value Command value
     * @return Future representing the command execution result
     */
    ListenableFuture<ProgramController> command(String name, Object value);
    
    /**
     * Stops the running program gracefully
     * @return Future representing the stop operation
     */
    ListenableFuture<ProgramController> stop();
    
    /**
     * Suspends the running program
     * @return Future representing the suspend operation
     */
    ListenableFuture<ProgramController> suspend();
    
    /**
     * Resumes a suspended program
     * @return Future representing the resume operation
     */
    ListenableFuture<ProgramController> resume();
    
    /**
     * Gets the current state of the program
     * @return Current program state
     */
    State getState();
    
    /**
     * Gets the program run ID
     * @return ProgramRunId for this execution
     */
    ProgramRunId getProgramRunId();
}

import co.cask.cdap.app.runtime.spark.Spark1ProgramRuntimeProvider;
import co.cask.cdap.proto.ProgramType;
import co.cask.cdap.app.runtime.ProgramRunner;

// Create runtime provider
Spark1ProgramRuntimeProvider provider = new Spark1ProgramRuntimeProvider();

// Verify it handles Spark programs
ProgramType supportedType = provider.getProgramType();
assert supportedType == ProgramType.SPARK;

// Check if program type is supported
boolean isSupported = provider.isSupported(ProgramType.SPARK, cConf);

// Create program runner
ProgramRunner runner = provider.createProgramRunner(ProgramType.SPARK, Mode.DISTRIBUTED, injector);

import co.cask.cdap.app.runtime.ProgramController;
import co.cask.cdap.app.runtime.ProgramController.State;

// Get program controller (typically from program runner execution)
ProgramController controller = // ... obtained from program execution

// Monitor program state
State currentState = controller.getState();
System.out.println("Program state: " + currentState);

// Send command to program
controller.command("checkpoint", "/path/to/checkpoint").get();

// Suspend program
controller.suspend().get();

// Resume program
controller.resume().get();

// Stop program gracefully
controller.stop().get();

/**
 * Enumeration of possible program states
 */
public enum State {
    STARTING,     // Program is initializing
    ALIVE,        // Program is running normally
    SUSPENDING,   // Program is being suspended
    SUSPENDED,    // Program is suspended
    RESUMING,     // Program is resuming from suspension
    STOPPING,     // Program is shutting down
    COMPLETED,    // Program completed successfully
    KILLED,       // Program was forcibly terminated
    ERROR         // Program encountered an error
}

/**
 * Program type enumeration
 */
public enum ProgramType {
    MAPREDUCE,   // MapReduce programs
    WORKFLOW,    // Workflow programs
    SERVICE,     // Service programs
    SPARK,       // Spark programs
    WORKER       // Worker programs
}

/**
 * Unique identifier for a program run
 */
public class ProgramRunId {
    /**
     * Gets the namespace containing the program
     * @return Namespace name
     */
    public String getNamespace();
    
    /**
     * Gets the application containing the program
     * @return Application name
     */
    public String getApplication();
    
    /**
     * Gets the program type
     * @return ProgramType of this program
     */
    public ProgramType getType();
    
    /**
     * Gets the program name
     * @return Program name
     */
    public String getProgram();
    
    /**
     * Gets the run ID
     * @return Unique run identifier
     */
    public String getRun();
}

/**
 * Configuration options for program execution
 */
public class ProgramOptions {
    /**
     * Gets runtime arguments for the program
     * @return Map of argument key-value pairs
     */
    public Map<String, String> getArguments();
    
    /**
     * Gets user-provided arguments
     * @return Map of user argument key-value pairs
     */
    public Map<String, String> getUserArguments();
    
    /**
     * Checks if debug mode is enabled
     * @return true if debug mode is enabled
     */
    public boolean isDebug();
    
    /**
     * Gets the program JAR location
     * @return Location of the program JAR file
     */
    public Location getProgramJarLocation();
}

/**
 * Interface for program runners that execute CDAP programs
 */
public interface ProgramRunner {
    /**
     * Runs a program with the specified options
     * @param programOptions Configuration for the program execution
     * @return ProgramController for managing the running program
     */
    ProgramController run(ProgramOptions programOptions);
}

/**
 * Execution mode enumeration for programs
 */
public enum Mode {
    LOCAL,        // Local execution mode
    DISTRIBUTED   // Distributed execution mode
}

/**
 * Guice injector interface for dependency injection
 */
public interface Injector {
    /**
     * Gets an instance of the specified type
     * @param type Class type to instantiate
     * @param <T> Type parameter
     * @return Instance of the specified type
     */
    <T> T getInstance(Class<T> type);
}