Communication Interfaces and Models

/**
 * Interface for communication from executor to admin server
 * Handles job callbacks and executor lifecycle management
 */
public interface AdminBiz {
    
    /**
     * Send job execution callback results to admin server
     * @param callbackParamList - List of job execution results
     * @return Response indicating callback processing status
     */
    ReturnT<String> callback(List<HandleCallbackParam> callbackParamList);
    
    /**
     * Register executor with admin server
     * @param registryParam - Executor registration information
     * @return Response indicating registration status
     */
    ReturnT<String> registry(RegistryParam registryParam);
    
    /**
     * Unregister executor from admin server
     * @param registryParam - Executor registration information to remove
     * @return Response indicating unregistration status
     */
    ReturnT<String> registryRemove(RegistryParam registryParam);
}

/**
 * Interface for communication from admin server to executor
 * Handles job triggering and executor monitoring
 */
public interface ExecutorBiz {
    
    /**
     * Heartbeat check to verify executor is alive
     * @return Response indicating executor status
     */
    ReturnT<String> beat();
    
    /**
     * Idle heartbeat check for specific job
     * @param idleBeatParam - Parameters for idle beat check
     * @return Response indicating job thread status
     */
    ReturnT<String> idleBeat(IdleBeatParam idleBeatParam);
    
    /**
     * Trigger job execution
     * @param triggerParam - Job execution parameters
     * @return Response indicating trigger acceptance status
     */
    ReturnT<String> run(TriggerParam triggerParam);
    
    /**
     * Kill running job execution
     * @param killParam - Parameters for job termination
     * @return Response indicating kill operation status
     */
    ReturnT<String> kill(KillParam killParam);
    
    /**
     * Query job execution logs
     * @param logParam - Parameters for log query
     * @return Log content and metadata
     */
    ReturnT<LogResult> log(LogParam logParam);
}

/**
 * Generic response wrapper for RPC calls
 * @param <T> Response data type
 */
public class ReturnT<T> implements Serializable {
    
    // Response code constants
    /**
     * Success response code
     */
    public static final int SUCCESS_CODE = 200;
    
    /**
     * Failure response code
     */
    public static final int FAIL_CODE = 500;
    
    // Predefined response instances
    /**
     * Predefined success response
     */
    public static final ReturnT<String> SUCCESS = new ReturnT<String>(null);
    
    /**
     * Predefined failure response
     */
    public static final ReturnT<String> FAIL = new ReturnT<String>(FAIL_CODE, null);
    
    // Constructors
    /**
     * Create success response with data
     * @param content Response data
     */
    public ReturnT(T content);
    
    /**
     * Create response with code and message
     * @param code Response code (200=success, 500=failure)
     * @param msg Response message
     */
    public ReturnT(int code, String msg);
    
    /**
     * Create empty response (defaults to success)
     */
    public ReturnT();
    
    // Accessors
    /**
     * Get response code
     * @return Response code (200=success, 500=failure)
     */
    public int getCode();
    
    /**
     * Set response code
     * @param code Response code to set
     */
    public void setCode(int code);
    
    /**
     * Get response message
     * @return Response message or null
     */
    public String getMsg();
    
    /**
     * Set response message
     * @param msg Response message to set
     */
    public void setMsg(String msg);
    
    /**
     * Get response content/data
     * @return Response data or null
     */
    public T getContent();
    
    /**
     * Set response content/data
     * @param content Response data to set
     */
    public void setContent(T content);
}

// Success response with data
ReturnT<String> successResult = new ReturnT<>("Operation completed successfully");

// Failure response with error message
ReturnT<String> failureResult = new ReturnT<>(ReturnT.FAIL_CODE, "Validation failed");

// Check response status
if (result.getCode() == ReturnT.SUCCESS_CODE) {
    String data = result.getContent();
    // Handle success
} else {
    String error = result.getMsg();
    // Handle failure
}

/**
 * Parameters for job trigger requests from admin to executor
 */
public class TriggerParam implements Serializable {
    
    /**
     * Job identifier
     */
    private int jobId;
    
    /**
     * Job handler name
     */
    private String executorHandler;
    
    /**
     * Job execution parameters
     */
    private String executorParams;
    
    /**
     * Blocking strategy when job is already running
     */
    private String executorBlockStrategy;
    
    /**
     * Job execution timeout in seconds
     */
    private int executorTimeout;
    
    /**
     * Log identifier for this execution
     */
    private long logId;
    
    /**
     * Job trigger timestamp
     */
    private long logDateTime;
    
    /**
     * Glue script type (BEAN, GLUE_GROOVY, etc.)
     */
    private String glueType;
    
    /**
     * Glue script source code
     */
    private String glueSource;
    
    /**
     * Glue script last update time
     */
    private long glueUpdatetime;
    
    /**
     * Broadcast index for broadcast jobs
     */
    private int broadcastIndex;
    
    /**
     * Total broadcast count for broadcast jobs
     */
    private int broadcastTotal;
    
    // Standard getters and setters for all fields
}

/**
 * Parameters for job execution callbacks from executor to admin
 */
public class HandleCallbackParam implements Serializable {
    
    /**
     * Log identifier matching the trigger request
     */
    private long logId;
    
    /**
     * Job execution timestamp
     */
    private long logDateTim;
    
    /**
     * Job execution result code (200=success, 500=fail, 502=timeout)
     */
    private int handleCode;
    
    /**
     * Job execution result message
     */
    private String handleMsg;
    
    // Constructors
    /**
     * Create callback parameter
     * @param logId Log identifier
     * @param logDateTim Execution timestamp
     * @param handleCode Result code
     * @param handleMsg Result message
     */
    public HandleCallbackParam(long logId, long logDateTim, int handleCode, String handleMsg);
    
    // Standard getters and setters
}

/**
 * Parameters for executor registration with admin server
 */
public class RegistryParam implements Serializable {
    
    /**
     * Registry type (EXECUTOR or ADMIN)
     */
    private String registryGroup;
    
    /**
     * Registry key (typically application name)
     */
    private String registryKey;
    
    /**
     * Registry value (executor address)
     */
    private String registryValue;
    
    // Constructors
    /**
     * Create registration parameter
     * @param registryGroup Registry type
     * @param registryKey Application name
     * @param registryValue Executor address
     */
    public RegistryParam(String registryGroup, String registryKey, String registryValue);
    
    // Standard getters and setters
}

/**
 * Parameters for idle heartbeat checks
 */
public class IdleBeatParam implements Serializable {
    
    /**
     * Job identifier to check
     */
    private int jobId;
    
    // Constructors and standard getters/setters
}

/**
 * Parameters for job kill requests
 */
public class KillParam implements Serializable {
    
    /**
     * Job identifier to terminate
     */
    private int jobId;
    
    // Constructors and standard getters/setters
}

/**
 * Parameters for log query requests
 */
public class LogParam implements Serializable {
    
    /**
     * Job trigger timestamp
     */
    private long logDateTim;
    
    /**
     * Log identifier
     */
    private long logId;
    
    /**
     * Starting line number for log retrieval
     */
    private int fromLineNum;
    
    // Constructors and standard getters/setters
}

/**
 * Response containing job execution log data
 */
public class LogResult implements Serializable {
    
    /**
     * Starting line number of returned content
     */
    private int fromLineNum;
    
    /**
     * Ending line number of returned content
     */
    private int toLineNum;
    
    /**
     * Log content as string
     */
    private String logContent;
    
    /**
     * Whether there are more log lines available
     */
    private boolean isEnd;
    
    // Constructors and standard getters/setters
}

// 1. Admin triggers job execution
TriggerParam triggerParam = new TriggerParam();
triggerParam.setJobId(123);
triggerParam.setExecutorHandler("myJobHandler");
triggerParam.setExecutorParams("{\"mode\":\"batch\"}");
triggerParam.setLogId(456);

ReturnT<String> triggerResponse = executorBiz.run(triggerParam);

// 2. Executor processes job and sends callback
HandleCallbackParam callbackParam = new HandleCallbackParam(
    456, // logId
    System.currentTimeMillis(),
    200, // success code
    "Job completed successfully"
);

List<HandleCallbackParam> callbackList = Arrays.asList(callbackParam);
ReturnT<String> callbackResponse = adminBiz.callback(callbackList);

// Register executor with admin server
RegistryParam registryParam = new RegistryParam(
    "EXECUTOR",                    // registry group
    "my-application",              // app name
    "http://192.168.1.100:9999"    // executor address
);

ReturnT<String> registryResponse = adminBiz.registry(registryParam);

// Unregister on shutdown
ReturnT<String> unregisterResponse = adminBiz.registryRemove(registryParam);

// General executor heartbeat
ReturnT<String> beatResponse = executorBiz.beat();

// Job-specific idle check
IdleBeatParam idleBeatParam = new IdleBeatParam();
idleBeatParam.setJobId(123);

ReturnT<String> idleBeatResponse = executorBiz.idleBeat(idleBeatParam);

// Query job execution logs
LogParam logParam = new LogParam();
logParam.setLogId(456);
logParam.setLogDateTim(System.currentTimeMillis());
logParam.setFromLineNum(1);

ReturnT<LogResult> logResponse = executorBiz.log(logParam);

if (logResponse.getCode() == ReturnT.SUCCESS_CODE) {
    LogResult logResult = logResponse.getContent();
    String logContent = logResult.getLogContent();
    boolean hasMoreLines = !logResult.isEnd();
    
    // Display or process log content
    System.out.println(logContent);
    
    // Fetch more lines if available
    if (hasMoreLines) {
        logParam.setFromLineNum(logResult.getToLineNum() + 1);
        // Query next batch...
    }
}

// Kill running job
KillParam killParam = new KillParam();
killParam.setJobId(123);

ReturnT<String> killResponse = executorBiz.kill(killParam);

if (killResponse.getCode() == ReturnT.SUCCESS_CODE) {
    System.out.println("Job killed successfully");
} else {
    System.out.println("Kill failed: " + killResponse.getMsg());
}

public void handleResponse(ReturnT<?> response) {
    switch (response.getCode()) {
        case ReturnT.SUCCESS_CODE:
            // Handle success
            System.out.println("Operation successful: " + response.getMsg());
            break;
            
        case ReturnT.FAIL_CODE:
            // Handle failure
            System.err.println("Operation failed: " + response.getMsg());
            break;
            
        default:
            // Handle custom codes
            System.out.println("Custom response code " + response.getCode() + ": " + response.getMsg());
            break;
    }
}

public ReturnT<String> callWithRetry(AdminBiz adminBiz, HandleCallbackParam callbackParam) {
    int maxRetries = 3;
    int retryDelay = 1000; // 1 second
    
    for (int attempt = 1; attempt <= maxRetries; attempt++) {
        try {
            List<HandleCallbackParam> callbacks = Arrays.asList(callbackParam);
            ReturnT<String> response = adminBiz.callback(callbacks);
            
            if (response.getCode() == ReturnT.SUCCESS_CODE) {
                return response;
            }
            
            System.out.println("Attempt " + attempt + " failed: " + response.getMsg());
            
        } catch (Exception e) {
            System.out.println("Attempt " + attempt + " error: " + e.getMessage());
        }
        
        if (attempt < maxRetries) {
            try {
                Thread.sleep(retryDelay * attempt); // Exponential backoff
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }
    
    return new ReturnT<>(ReturnT.FAIL_CODE, "All retry attempts failed");
}