JVMCBC-1674 check before null for all access of numFailedRefreshes and registrations

Change-Id: I1c084b675ef763d2e598bc225e7ab493dfdbba74
Reviewed-on: https://review.couchbase.org/c/couchbase-jvm-clients/+/233564
Tested-by: Build Bot <[email protected]>
Reviewed-by: David Nault <[email protected]>
(cherry picked from commit c3e0ba6)
Reviewed-on: https://review.couchbase.org/c/couchbase-jvm-clients/+/234331
Tested-by: Saurabh Mishra <[email protected]>

Author: CosmicSaaurabh

core-io/src/main/java/com/couchbase/client/core/config/refresher/KeyValueBucketRefresher.java

@@ -159,6 +159,20 @@ protected Duration pollerInterval() {
     return POLLER_INTERVAL;
   }
 
+  private void incrementFailedRefreshes(String name) {
+    AtomicInteger counter = numFailedRefreshes.get(name);
+    if (counter != null) {
+      counter.incrementAndGet();
+    }
+  }
+
+  private void clearFailedRefreshes(String name) {
+    AtomicInteger counter = numFailedRefreshes.get(name);
+    if (counter != null) {
+      counter.set(0);
+    }
+  }
+
   /**
    * Helper method to make sure the config poll interval is set to a fixed bound every time.
    *
@@ -201,7 +215,8 @@ private Mono<ProposedBucketConfigContext> maybeUpdateBucket(final String name, b
 
     if (allowed) {
       List<NodeInfo> nodes = filterEligibleNodes(name);
-      if (numFailedRefreshes.get(name).get() >= nodes.size()) {
+      AtomicInteger counter = numFailedRefreshes.get(name);
+      if (counter != null && counter.get() >= nodes.size()) {
         provider.signalConfigRefreshFailed(ConfigRefreshFailure.ALL_NODES_TRIED_ONCE_WITHOUT_SUCCESS);
         numFailedRefreshes.get(name).set(0);
       }
@@ -274,7 +289,7 @@ private Flux<ProposedBucketConfigContext> fetchConfigPerNode(final String name,
         .wrap(request, request.response(), true)
         .filter(response -> {
           if (!response.status().success()) {
-            numFailedRefreshes.get(name).incrementAndGet();
+            incrementFailedRefreshes(name);
             eventBus.publish(new BucketConfigRefreshFailedEvent(
               core.context(),
               BucketConfigRefreshFailedEvent.RefresherType.KV,
@@ -287,9 +302,9 @@ private Flux<ProposedBucketConfigContext> fetchConfigPerNode(final String name,
         .map(response ->
           new ProposedBucketConfigContext(name, new String(response.content(), UTF_8), nodeInfo.hostname())
         )
-        .doOnSuccess(r -> numFailedRefreshes.get(name).set(0))
+        .doOnSuccess(r -> clearFailedRefreshes(name))
         .onErrorResume(t -> {
-          numFailedRefreshes.get(name).incrementAndGet();
+          incrementFailedRefreshes(name);
           eventBus.publish(new BucketConfigRefreshFailedEvent(
             core.context(),
             BucketConfigRefreshFailedEvent.RefresherType.KV,