private void initConnection() throws Exception {
        LOGGER.info("Initializing Inlong Manager AirflowConnection for Airflow ... ");
        // Check if Airflow has the Inlong AirflowConnection
        AirflowResponse<AirflowConnection> response = serverClient.sendRequest(
                new AirflowConnectionGetter(airflowConfig.getConnectionId()));
        if (!response.isSuccess()) {
            AirflowConnection newConn = new AirflowConnection(airflowConfig.getConnectionId(), "HTTP", "",
                    airflowConfig.getInlongManagerHost(), airflowConfig.getInlongUsername(), SUBMIT_OFFLINE_JOB_URI,
                    airflowConfig.getInlongManagerPort(), airflowConfig.getInlongPassword(), "");
            response = serverClient.sendRequest(new AirflowConnectionCreator(newConn));
            LOGGER.info("AirflowConnection registration response: {}", response.toString());
            if (!response.isSuccess()) {
                LOGGER.error("Initialization connection failed.");
                throw new AirflowScheduleException(INIT_CONNECTION_FAILED, "Initialization connection failed.");
            }
        }
    }
Response for processing the start/register/unregister/update/stop requests from {@link AirflowScheduleClient}
inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/airflow/AirflowScheduleEngine.java