Use keyspace metadata in BasicLoadBalancingPolicy#getReplicas

Modifies the getReplicas method to do the tablet lookup if and only if the
keyspace metadata indicates that it's a tablets-based keyspace. Otherwise refer
to the token map.
Previous behavior was to try tablet map lookup first regardless of the
keyspace configuration.

Author: Bouncheck

core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/BasicLoadBalancingPolicy.java

@@ -37,6 +37,7 @@
 import com.datastax.oss.driver.api.core.metadata.Tablet;
 import com.datastax.oss.driver.api.core.metadata.TabletMap;
 import com.datastax.oss.driver.api.core.metadata.TokenMap;
+import com.datastax.oss.driver.api.core.metadata.schema.KeyspaceMetadata;
 import com.datastax.oss.driver.api.core.metadata.token.Partitioner;
 import com.datastax.oss.driver.api.core.metadata.token.Token;
 import com.datastax.oss.driver.api.core.session.Request;
@@ -328,7 +329,12 @@ protected Set<Node> getReplicas(@Nullable Request request, @Nullable Session ses
       return Collections.emptySet();
     }
 
-    if (table != null) {
+    Optional<KeyspaceMetadata> ksMetadata =
+        context.getMetadataManager().getMetadata().getKeyspace(keyspace);
+    if (ksMetadata.isPresent() && ksMetadata.get().isUsingTablets()) {
+      if (table == null) {
+        return Collections.emptySet();
+      }
       if (token == null) {
         if (partitioner != null) {
           token = partitioner.hash(key);
@@ -338,12 +344,10 @@ protected Set<Node> getReplicas(@Nullable Request request, @Nullable Session ses
         Tablet targetTablet =
             tabletMap.getTablet(keyspace, table, ((TokenLong64) token).getValue());
         if (targetTablet != null) {
-          Set<Node> replicas = targetTablet.getReplicaNodes();
-          if (!replicas.isEmpty()) {
-            return replicas;
-          }
+          return targetTablet.getReplicaNodes();
         }
       }
+      return Collections.emptySet();
     }
 
     if (!maybeTokenMap.isPresent()) {

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/schema/queries/CassandraSchemaQueries.java

@@ -213,17 +213,20 @@ private void handleResult(
       AdminResult result,
       Throwable error,
       Function<Iterable<AdminRow>, CassandraSchemaRows.Builder> builderUpdater,
-      boolean ignoreTargetDoesNotExistErrors) {
+      boolean ignoreServerErrors) {
 
     // If another query already failed, we've already propagated the failure so just ignore this one
     if (schemaRowsFuture.isCompletedExceptionally()) {
       return;
     }
 
-    if (ignoreTargetDoesNotExistErrors && error instanceof UnexpectedResponseException) {
+    // Meant to allow through "(keyspace/table) does not exist" or "unconfigured" errors for
+    // specific, optional queries
+    if (ignoreServerErrors && error instanceof UnexpectedResponseException) {
       UnexpectedResponseException castedError = (UnexpectedResponseException) error;
-      if (castedError.message.opcode == ProtocolConstants.ErrorCode.SERVER_ERROR
-          && castedError.getMessage().contains("does not exist")) {
+      if (castedError.message.opcode == ProtocolConstants.ErrorCode.SERVER_ERROR) {
+        LOG.debug("Silencing error: ", error);
+        // Consider such query 'done', but ignore its result
         pendingQueries -= 1;
         if (pendingQueries == 0) {
           LOG.debug(
@@ -244,8 +247,7 @@ private void handleResult(
             .nextPage()
             .whenCompleteAsync(
                 (nextResult, nextError) ->
-                    handleResult(
-                        nextResult, nextError, builderUpdater, ignoreTargetDoesNotExistErrors),
+                    handleResult(nextResult, nextError, builderUpdater, ignoreServerErrors),
                 adminExecutor);
       } else {
         pendingQueries -= 1;