Updates to resolver

Author: Tim Middleton

coherence/common.go

@@ -44,6 +44,9 @@ const (
 	// envResolverDebug enables resolver debug messages to be displayed.
 	envResolverDebug = "COHERENCE_RESOLVER_DEBUG"
 
+	// envResolverDebug sets the number of retries when the resolver fails.
+	envResolverRetries = "COHERENCE_RESOLVER_RETRIES"
+
 	// envResolverDebug enables randomization of addresses returned by resolver
 	envResolverRandomize = "COHERENCE_RESOLVER_RANDOMIZE"
 

coherence/event.go

@@ -1219,7 +1219,7 @@ func reRegisterListeners[K comparable, V any](ctx context.Context, namedMap *Nam
 			bc.filterListenersV1 = make(map[filters.Filter]*listenerGroupV1[K, V], 0)
 			bc.filterIDToGroupV1 = make(map[int64]*listenerGroupV1[K, V], 0)
 
-			// re-ensure all the caches as the connected has gone and so has the gRPC Proxy
+			// re-ensure all the caches as the connection has gone and so has the gRPC Proxy
 			for _, c := range cacheNames {
 				cacheID, err3 := bc.session.v1StreamManagerCache.ensureCache(context.Background(), c)
 				if err3 != nil {

coherence/resolver.go

@@ -12,13 +12,16 @@ import (
 	"github.com/oracle/coherence-go-client/v2/coherence/discovery"
 	"google.golang.org/grpc/resolver"
 	"math/rand"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
 )
 
 const (
-	nsLookupScheme = "coherence"
+	nsLookupScheme       = "coherence"
+	defaultRetries       = 20
+	defaultResolverDelay = 1000 // ms
 )
 
 var (
@@ -42,16 +45,27 @@ func (b *nsLookupResolverBuilder) Build(target resolver.Target, cc resolver.Clie
 	}
 	checkResolverDebug()
 
+	// set the number of resolver retried
+	retries := getStringValueFromEnvVarOrDefault(envResolverRetries, "20")
+	retriesValue, err := strconv.Atoi(retries)
+	if err != nil {
+		retriesValue = defaultRetries
+	}
+
+	resolverDebug("resolver retries=%v", retriesValue)
+	r.resolverRetries = retriesValue
+
 	r.start()
 	return r, nil
 }
 func (*nsLookupResolverBuilder) Scheme() string { return nsLookupScheme }
 
 type nsLookupResolver struct {
-	target    resolver.Target
-	cc        resolver.ClientConn
-	mutex     sync.Mutex
-	addrStore map[string][]string
+	target          resolver.Target
+	cc              resolver.ClientConn
+	mutex           sync.Mutex
+	addrStore       map[string][]string
+	resolverRetries int
 }
 
 func (r *nsLookupResolver) resolve() {
@@ -60,10 +74,10 @@ func (r *nsLookupResolver) resolve() {
 	defer r.mutex.Unlock()
 
 	if len(grpcEndpoints) == 0 {
-		// try 8 times over 2 seconds to get gRPC addresses as we may be in the middle of fail-over
-		for i := 0; i < 8; i++ {
-			resolverDebug("retrying NSLookup attempt", i)
-			time.Sleep(time.Duration(250) * time.Millisecond)
+		// try r.resolverRetries; times over 2 seconds to get gRPC addresses as we may be in the middle of fail-over
+		for i := 1; i <= r.resolverRetries; i++ {
+			resolverDebug("retrying NSLookup attempt: %v", i)
+			time.Sleep(time.Duration(defaultResolverDelay) * time.Millisecond)
 			grpcEndpoints = generateNSAddresses(r.target.Endpoint())
 			if len(grpcEndpoints) != 0 {
 				break

coherence/session.go

@@ -19,6 +19,7 @@ import (
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/resolver"
+	"google.golang.org/grpc/status"
 	"log"
 	"os"
 	"reflect"
@@ -510,7 +511,14 @@ func (s *Session) ensureConnection() error {
 		s.v1StreamManagerCache = manager
 		apiMessage = fmt.Sprintf(" %v", manager)
 	} else {
-		s.debug("error connecting to session via v1, falling back to v0: %v", err1)
+		// check if this is a gRPC status error
+		if sts, ok := status.FromError(err1); ok {
+			if sts.Message() == "Method not found: coherence.proxy.v1.ProxyService/subChannel" {
+				s.debug("error connecting to session via v1, falling back to v0: %v", err1)
+			} else {
+				s.debug("received a different gRPC error: %v", err1)
+			}
+		}
 	}
 
 	logMessage(INFO, "Session [%s] connected to [%s]%s", s.sessionID, s.sessOpts.Address, apiMessage)
@@ -547,14 +555,14 @@ func (s *Session) ensureConnection() error {
 				return
 			}
 
-			if newState == connectivity.Ready || newState == connectivity.Idle {
+			if newState == connectivity.Ready { //|| newState == connectivity.Idle {
 				if !firstConnect && !connected {
-					// Reconnect
+					// Reconnected
 					disconnectTime = 0
 					session.closed = false
 					connected = true
 
-					logMessage(INFO, "Session [%s] re-connected to address %s", session.sessionID, session.sessOpts.Address)
+					logMessage(INFO, "Session [%s] re-connected to address %s (%v)", session.sessionID, session.sessOpts.Address, newState)
 					session.dispatch(Reconnected, func() SessionLifecycleEvent {
 						return newSessionLifecycleEvent(session, Reconnected)
 					})

coherence/session_test.go

@@ -1,5 +1,5 @@
 /*
-* Copyright (c) 2022, 2023 Oracle and/or its affiliates.
+* Copyright (c) 2022, 2024 Oracle and/or its affiliates.
 * Licensed under the Universal Permissive License v 1.0 as shown at
 * https://oss.oracle.com/licenses/upl.
  */
@@ -9,6 +9,7 @@ package coherence
 import (
 	"context"
 	"github.com/onsi/gomega"
+	"os"
 	"strconv"
 	"testing"
 	"time"
@@ -21,6 +22,8 @@ func TestSessionValidation(t *testing.T) {
 		ctx = context.Background()
 	)
 
+	os.Setenv("COHERENCE_SESSION_DEBUG", "true")
+
 	_, err = NewSession(ctx, WithFormat("not-json"))
 	g.Expect(err).To(gomega.Equal(ErrInvalidFormat))
 

java/coherence-go-queues/pom.xml

@@ -50,10 +50,7 @@
     <profile>
       <id>javax</id>
       <activation>
-        <!-- This is a work-around for the fact that activeByDefault does not do what you'd think it should -->
-        <file>
-          <exists>.</exists>
-        </file>
+        <activeByDefault>false</activeByDefault>
       </activation>
       <dependencies>
         <dependency>

scripts/run-checkin-test.sh

@@ -0,0 +1,19 @@
+#!/bin/bash
+
+#
+# Copyright (c) 2022, 2024 Oracle and/or its affiliates.
+# Licensed under the Universal Permissive License v 1.0 as shown at
+# https://oss.oracle.com/licenses/upl.
+#
+
+# This script runs some tests that should succeed to be sure we can push
+set -e
+
+echo "Coherence CE 24.09 All Tests gRPC v1"
+COHERENCE_BASE_IMAGE=gcr.io/distroless/java17 PROFILES=,jakarta,-javax COHERENCE_VERSION=24.09 make clean generate-proto generate-proto-v1 build-test-images test-e2e-standalone
+
+echo "Coherence CE 24.09 with queues"
+COHERENCE_BASE_IMAGE=gcr.io/distroless/java17 PROFILES=,jakarta,-javax,queues COHERENCE_VERSION=24.09 make clean generate-proto generate-proto-v1 build-test-images test-e2e-standalone-queues
+
+echo "Coherence CE 22.06.10"
+COHERENCE_VERSION=22.06.10 PROFILES=,-jakarta,javax make clean generate-proto build-test-images test-e2e-standalone