JAR Management

/**
 * Handler for uploading JAR files to the Flink web server.
 * Supports multipart/form-data uploads and stores JARs in the configured directory.
 */
public class JarUploadHandler extends AbstractRestHandler<RestfulGateway, EmptyRequestBody, JarUploadResponseBody, EmptyMessageParameters> {
    /**
     * Create a new JAR upload handler.
     * 
     * @param leaderRetriever Gateway retriever for accessing Flink cluster
     * @param timeout Request timeout for upload operations
     * @param responseHeaders HTTP headers to include in responses
     * @param messageHeaders Message header specification for this endpoint
     * @param jarDir Directory where uploaded JARs will be stored
     * @param executor Executor for handling upload operations
     */
    public JarUploadHandler(
        GatewayRetriever<? extends RestfulGateway> leaderRetriever,
        Time timeout,
        Map<String, String> responseHeaders,
        MessageHeaders<EmptyRequestBody, JarUploadResponseBody, EmptyMessageParameters> messageHeaders,
        Path jarDir,
        Executor executor
    );
}

/**
 * Handler for listing uploaded JAR files and their available entry points.
 * Returns detailed information about each JAR including main classes and program arguments.
 */
public class JarListHandler extends AbstractRestHandler<RestfulGateway, EmptyRequestBody, JarListInfo, EmptyMessageParameters> {
    /**
     * Create a new JAR list handler.
     * 
     * @param leaderRetriever Gateway retriever for accessing Flink cluster
     * @param timeout Request timeout for listing operations
     * @param responseHeaders HTTP headers to include in responses
     * @param messageHeaders Message header specification for this endpoint
     * @param localAddressFuture Future containing the local server address
     * @param jarDir Directory containing uploaded JARs
     * @param configuration Flink configuration
     * @param executor Executor for JAR analysis operations
     */
    public JarListHandler(
        GatewayRetriever<? extends RestfulGateway> leaderRetriever,
        Time timeout,
        Map<String, String> responseHeaders,
        MessageHeaders<EmptyRequestBody, JarListInfo, EmptyMessageParameters> messageHeaders,
        CompletableFuture<String> localAddressFuture,
        File jarDir,
        Configuration configuration,
        Executor executor
    );
}

/**
 * Handler for executing uploaded JAR files as Flink jobs.
 * Supports job configuration including parallelism, savepoints, and program arguments.
 */
public class JarRunHandler extends AbstractRestHandler<DispatcherGateway, JarRunRequestBody, JarRunResponseBody, JarRunMessageParameters> {
    /**
     * Create a new JAR run handler.
     * 
     * @param leaderRetriever Gateway retriever for accessing Flink cluster
     * @param timeout Request timeout for job submission
     * @param responseHeaders HTTP headers to include in responses
     * @param messageHeaders Message header specification for this endpoint
     * @param jarDir Directory containing uploaded JARs
     * @param configuration Flink configuration
     * @param executor Executor for job submission operations
     * @param applicationRunnerSupplier Supplier for creating application runners
     */
    public JarRunHandler(
        GatewayRetriever<? extends DispatcherGateway> leaderRetriever,
        Time timeout,
        Map<String, String> responseHeaders,
        MessageHeaders<JarRunRequestBody, JarRunResponseBody, JarRunMessageParameters> messageHeaders,
        Path jarDir,
        Configuration configuration,
        Executor executor,
        Supplier<ApplicationRunner> applicationRunnerSupplier
    );
}

/**
 * Handler for deleting uploaded JAR files from the server.
 * Removes both the JAR file and any associated metadata.
 */
public class JarDeleteHandler extends AbstractRestHandler<RestfulGateway, EmptyRequestBody, EmptyResponseBody, JarDeleteMessageParameters> {
    /**
     * Create a new JAR delete handler.
     * 
     * @param leaderRetriever Gateway retriever for accessing Flink cluster
     * @param timeout Request timeout for delete operations
     * @param responseHeaders HTTP headers to include in responses
     * @param messageHeaders Message header specification for this endpoint
     * @param jarDir Directory containing uploaded JARs
     * @param executor Executor for delete operations
     */
    public JarDeleteHandler(
        GatewayRetriever<? extends RestfulGateway> leaderRetriever,
        Time timeout,
        Map<String, String> responseHeaders,
        MessageHeaders<EmptyRequestBody, EmptyResponseBody, JarDeleteMessageParameters> messageHeaders,
        Path jarDir,
        Executor executor
    );
}

/**
 * Handler for generating execution plans from uploaded JAR files.
 * Shows the job graph that would be executed without actually running the job.
 */
public class JarPlanHandler extends AbstractRestHandler<RestfulGateway, JarPlanRequestBody, JobPlanInfo, JarPlanMessageParameters> {
    /**
     * Create a new JAR plan handler with default plan generator.
     * 
     * @param leaderRetriever Gateway retriever for accessing Flink cluster
     * @param timeout Request timeout for plan generation
     * @param responseHeaders HTTP headers to include in responses
     * @param messageHeaders Message header specification for this endpoint
     * @param jarDir Directory containing uploaded JARs
     * @param configuration Flink configuration
     * @param executor Executor for plan generation operations
     */
    public JarPlanHandler(
        GatewayRetriever<? extends RestfulGateway> leaderRetriever,
        Time timeout,
        Map<String, String> responseHeaders,
        MessageHeaders<JarPlanRequestBody, JobPlanInfo, JarPlanMessageParameters> messageHeaders,
        Path jarDir,
        Configuration configuration,
        Executor executor
    );
    
    /**
     * Create a new JAR plan handler with custom plan generator.
     * 
     * @param leaderRetriever Gateway retriever for accessing Flink cluster
     * @param timeout Request timeout for plan generation
     * @param responseHeaders HTTP headers to include in responses
     * @param messageHeaders Message header specification for this endpoint
     * @param jarDir Directory containing uploaded JARs
     * @param configuration Flink configuration
     * @param executor Executor for plan generation operations
     * @param planGenerator Custom function to generate JobPlanInfo from JobGraph
     */
    public JarPlanHandler(
        GatewayRetriever<? extends RestfulGateway> leaderRetriever,
        Time timeout,
        Map<String, String> responseHeaders,
        MessageHeaders<JarPlanRequestBody, JobPlanInfo, JarPlanMessageParameters> messageHeaders,
        Path jarDir,
        Configuration configuration,
        Executor executor,
        Function<JobGraph, JobPlanInfo> planGenerator
    );
}

import org.apache.flink.runtime.webmonitor.handlers.*;
import org.apache.flink.configuration.Configuration;
import java.nio.file.Paths;

// Setup configuration
Configuration config = new Configuration();
Path jarDir = Paths.get("/tmp/flink-jars");
Time timeout = Time.seconds(30);
Map<String, String> headers = Collections.emptyMap();

// Create handlers
JarUploadHandler uploadHandler = new JarUploadHandler(
    leaderRetriever,
    timeout,
    headers,
    JarUploadHeaders.getInstance(),
    jarDir,
    executor
);

JarListHandler listHandler = new JarListHandler(
    leaderRetriever,
    timeout,
    headers,
    JarListHeaders.getInstance(),
    jarDir,
    config,
    executor
);

JarRunHandler runHandler = new JarRunHandler(
    leaderRetriever,
    timeout,
    headers,
    JarRunHeaders.getInstance(),
    jarDir,
    config,
    executor,
    applicationRunnerSupplier
);

// Upload a JAR (would typically be handled by HTTP multipart upload)
// The actual upload is handled by the JarUploadHandler via HTTP POST

// List uploaded JARs
HandlerRequest<EmptyRequestBody, EmptyMessageParameters> listRequest = 
    HandlerRequest.create(EmptyRequestBody.getInstance(), EmptyMessageParameters.getInstance());
CompletableFuture<JarListInfo> jarList = listHandler.handleRequest(listRequest, restfulGateway);

// Run a JAR
JarRunRequestBody runRequest = new JarRunRequestBody();
runRequest.setEntryClassName("com.example.MyFlinkJob");
runRequest.setProgramArguments(Arrays.asList("--input", "/path/to/input"));
runRequest.setParallelism(4);

HandlerRequest<JarRunRequestBody, JarRunMessageParameters> runHandlerRequest =
    HandlerRequest.create(runRequest, jarRunMessageParameters, pathParameters, queryParameters);
CompletableFuture<JarRunResponseBody> jobResult = runHandler.handleRequest(runHandlerRequest, restfulGateway);

// Typical error handling pattern in handler usage
jarRunHandler.handleRequest(request, gateway)
    .whenComplete((result, throwable) -> {
        if (throwable != null) {
            if (throwable instanceof NotFoundException) {
                // Handle JAR not found
            } else if (throwable instanceof BadRequestException) {
                // Handle invalid request parameters
            } else {
                // Handle other errors
            }
        } else {
            // Process successful result
            JobID jobId = result.getJobId();
        }
    });