public static MesosConfiguration createMesosSchedulerConfiguration ( Configuration flinkConfig , String hostname ) { Protos . FrameworkInfo . Builder frameworkInfo = Protos . FrameworkInfo . newBuilder ( ) . setHostname ( hostname ) ; Protos . Credential . Builder credential = null ; if ( ! flinkConfig . contains ( MesosOptions . MASTER_URL ) ) { throw new IllegalConfigurationException ( MesosOptions . MASTER_URL . key ( ) + " must be configured." ) ; } String masterUrl = flinkConfig . getString ( MesosOptions . MASTER_URL ) ; Duration failoverTimeout = FiniteDuration . apply ( flinkConfig . getInteger ( MesosOptions . FAILOVER_TIMEOUT_SECONDS ) , TimeUnit . SECONDS ) ; frameworkInfo . setFailoverTimeout ( failoverTimeout . toSeconds ( ) ) ; frameworkInfo . setName ( flinkConfig . getString ( MesosOptions . RESOURCEMANAGER_FRAMEWORK_NAME ) ) ; frameworkInfo . setRole ( flinkConfig . getString ( MesosOptions . RESOURCEMANAGER_FRAMEWORK_ROLE ) ) ; frameworkInfo . setUser ( flinkConfig . getString ( MesosOptions . RESOURCEMANAGER_FRAMEWORK_USER ) ) ; if ( flinkConfig . contains ( MesosOptions . RESOURCEMANAGER_FRAMEWORK_PRINCIPAL ) ) { frameworkInfo . setPrincipal ( flinkConfig . getString ( MesosOptions . RESOURCEMANAGER_FRAMEWORK_PRINCIPAL ) ) ; credential = Protos . Credential . newBuilder ( ) ; credential . setPrincipal ( frameworkInfo . getPrincipal ( ) ) ; // some environments use a side-channel to communicate the secret to Mesos, // and thus don't set the 'secret' configuration setting if ( flinkConfig . contains ( MesosOptions . RESOURCEMANAGER_FRAMEWORK_SECRET ) ) { credential . setSecret ( flinkConfig . getString ( MesosOptions . RESOURCEMANAGER_FRAMEWORK_SECRET ) ) ; } } MesosConfiguration mesos = new MesosConfiguration ( masterUrl , frameworkInfo , scala . Option . apply ( credential ) ) ; return mesos ; }
Loads and validates the Mesos scheduler configuration .