@@ -359,75 +359,53 @@ public void startThreads() throws IOException {
359359    try  {
360360      keyCache  = CuratorCache .bridgeBuilder (zkClient , ZK_DTSM_MASTER_KEY_ROOT )
361361          .build ();
362-       if  (keyCache  != null ) {
363-         CuratorCacheListener  keyCacheListener  = CuratorCacheListener .builder ()
364-             .forCreates (childData  -> {
362+       CuratorCacheListener  keyCacheListener  = CuratorCacheListener .builder ()
363+           .forCreatesAndChanges ((oldNode , node ) -> {
364+             try  {
365+               processKeyAddOrUpdate (node .getData ());
366+             } catch  (IOException  e ) {
367+               LOG .error ("Error while processing Curator keyCacheListener " 
368+                   + "NODE_CREATED / NODE_CHANGED event" );
369+               throw  new  UncheckedIOException (e );
370+             }
371+           })
372+           .forDeletes (childData  -> processKeyRemoved (childData .getPath ()))
373+           .build ();
374+       keyCache .listenable ().addListener (keyCacheListener );
375+       keyCache .start ();
376+       loadFromZKCache (false );
377+     } catch  (Exception  e ) {
378+       throw  new  IOException ("Could not start Curator keyCacheListener for keys" ,
379+           e );
380+     }
381+     if  (isTokenWatcherEnabled ) {
382+       LOG .info ("TokenCache is enabled" );
383+       try  {
384+         tokenCache  = CuratorCache .bridgeBuilder (zkClient , ZK_DTSM_TOKENS_ROOT )
385+             .build ();
386+         CuratorCacheListener  tokenCacheListener  = CuratorCacheListener .builder ()
387+             .forCreatesAndChanges ((oldNode , node ) -> {
365388              try  {
366-                 processKeyAddOrUpdate ( childData .getData ());
389+                 processTokenAddOrUpdate ( node .getData ());
367390              } catch  (IOException  e ) {
368-                 LOG .error ("Error while processing Curator keyCacheListener  " 
369-                     + "NODE_CREATED event" );
391+                 LOG .error ("Error while processing Curator tokenCacheListener  " 
392+                     + "NODE_CREATED / NODE_CHANGED  event" );
370393                throw  new  UncheckedIOException (e );
371394              }
372395            })
373-             .forChanges (( oldNode ,  node )  -> {
396+             .forDeletes ( childData  -> {
374397              try  {
375-                 processKeyAddOrUpdate ( node . getData () );
398+                 processTokenRemoved ( childData );
376399              } catch  (IOException  e ) {
377-                 LOG .error ("Error while processing Curator keyCacheListener  " 
378-                     + "NODE_CHANGED  event" );
400+                 LOG .error ("Error while processing Curator tokenCacheListener  " 
401+                     + "NODE_DELETED  event" );
379402                throw  new  UncheckedIOException (e );
380403              }
381404            })
382-             .forDeletes (childData  -> processKeyRemoved (childData .getPath ()))
383405            .build ();
384-         keyCache .listenable ().addListener (keyCacheListener );
385-         keyCache .start ();
386-         loadFromZKCache (false );
387-       }
388-     } catch  (Exception  e ) {
389-       throw  new  IOException ("Could not start Curator keyCacheListener for keys" ,
390-           e );
391-     }
392-     if  (isTokenWatcherEnabled ) {
393-       LOG .info ("TokenCache is enabled" );
394-       try  {
395-         tokenCache  = CuratorCache .bridgeBuilder (zkClient , ZK_DTSM_TOKENS_ROOT )
396-             .build ();
397-         if  (tokenCache  != null ) {
398-           CuratorCacheListener  tokenCacheListener  = CuratorCacheListener .builder ()
399-               .forCreates (childData  -> {
400-                 try  {
401-                   processTokenAddOrUpdate (childData .getData ());
402-                 } catch  (IOException  e ) {
403-                   LOG .error ("Error while processing Curator tokenCacheListener " 
404-                       + "NODE_CREATED event" );
405-                   throw  new  UncheckedIOException (e );
406-                 }
407-               })
408-               .forChanges ((oldNode , node ) -> {
409-                 try  {
410-                   processTokenAddOrUpdate (node .getData ());
411-                 } catch  (IOException  e ) {
412-                   LOG .error ("Error while processing Curator tokenCacheListener " 
413-                       + "NODE_CHANGED event" );
414-                   throw  new  UncheckedIOException (e );
415-                 }
416-               })
417-               .forDeletes (childData  -> {
418-                 try  {
419-                   processTokenRemoved (childData );
420-                 } catch  (IOException  e ) {
421-                   LOG .error ("Error while processing Curator tokenCacheListener " 
422-                       + "NODE_DELETED event" );
423-                   throw  new  UncheckedIOException (e );
424-                 }
425-               })
426-               .build ();
427-           tokenCache .listenable ().addListener (tokenCacheListener );
428-           tokenCache .start ();
429-           loadFromZKCache (true );
430-         }
406+         tokenCache .listenable ().addListener (tokenCacheListener );
407+         tokenCache .start ();
408+         loadFromZKCache (true );
431409      } catch  (Exception  e ) {
432410        throw  new  IOException (
433411            "Could not start Curator tokenCacheListener for tokens" , e );
0 commit comments