public void run ( ) { try { log . info ( "{}#Start listing and watching..." , apiTypeClass ) ; ApiListType list = listerWatcher . list ( new CallGeneratorParams ( Boolean . FALSE , null , null ) ) ; V1ListMeta listMeta = Reflect . listMetadata ( list ) ; String resourceVersion = listMeta . getResourceVersion ( ) ; List < ApiType > items = Reflect . getItems ( list ) ; if ( log . isDebugEnabled ( ) ) { log . debug ( "{}#Extract resourceVersion {} list meta" , apiTypeClass , resourceVersion ) ; } this . syncWith ( items , resourceVersion ) ; this . lastSyncResourceVersion = resourceVersion ; if ( log . isDebugEnabled ( ) ) { log . debug ( "{}#Start watching with {}..." , apiTypeClass , lastSyncResourceVersion ) ; } while ( true ) { if ( ! isActive . get ( ) ) { if ( watch != null ) { watch . close ( ) ; return ; } } try { if ( log . isDebugEnabled ( ) ) { log . debug ( "{}#Start watch with resource version {}" , apiTypeClass , lastSyncResourceVersion ) ; } watch = listerWatcher . watch ( new CallGeneratorParams ( Boolean . TRUE , lastSyncResourceVersion , Long . valueOf ( Duration . ofMinutes ( 5 ) . toMillis ( ) ) . intValue ( ) ) ) ; watchHandler ( watch ) ; } catch ( Throwable t ) { log . info ( "{}#Watch connection get exception {}" , apiTypeClass , t . getMessage ( ) ) ; Throwable cause = t . getCause ( ) ; // If this is "connection refused" error, it means that most likely apiserver is not // responsive. // It doesn't make sense to re-list all objects because most likely we will be able to // restart // watch where we ended. // If that's the case wait and resend watch request. if ( cause != null && ( cause instanceof ConnectException ) ) { log . info ( "{}#Watch get connect exception, retry watch" , apiTypeClass ) ; Thread . sleep ( 1000L ) ; continue ; } if ( ( t instanceof RuntimeException ) && t . getMessage ( ) . contains ( "IO Exception during hasNext" ) ) { log . info ( "{}#Read timeout retry list and watch" , apiTypeClass ) ; return ; } log . error ( "{}#Watch failed as {} unexpected" , apiTypeClass , t . getMessage ( ) , t ) ; return ; } finally { if ( watch != null ) { watch . close ( ) ; watch = null ; } } } } catch ( Throwable t ) { log . error ( "{}#Failed to list-watch: {}" , apiTypeClass , t ) ; } }
run first lists all items and get the resource version at the moment of call and then use the resource version to watch .