issue 386 - handle cluster token overlap issue gracefully

msmygit · msmygit · commit 5f03be12fc11 · 2025-09-08T07:39:39.000-04:00
diff --git a/src/main/java/com/datastax/cdm/job/AbstractJobSession.java b/src/main/java/com/datastax/cdm/job/AbstractJobSession.java
@@ -30,6 +30,7 @@
 import com.datastax.cdm.job.IJobSessionFactory.JobType;
 import com.datastax.cdm.properties.KnownProperties;
 import com.datastax.cdm.properties.PropertyHelper;
+import com.datastax.cdm.schema.ClusterConfigurationException;
 import com.datastax.cdm.schema.CqlTable;
 import com.datastax.oss.driver.api.core.CqlSession;
 import com.datastax.oss.driver.shaded.guava.common.util.concurrent.RateLimiter;
@@ -55,45 +56,53 @@ protected AbstractJobSession(CqlSession originSession, CqlSession targetSession,
             return;
         }
 
-        rateLimiterOrigin = RateLimiter.create(propertyHelper.getInteger(KnownProperties.PERF_RATELIMIT_ORIGIN));
-        rateLimiterTarget = RateLimiter.create(propertyHelper.getInteger(KnownProperties.PERF_RATELIMIT_TARGET));
-
-        logger.info("PARAM -- Origin Rate Limit: {}", rateLimiterOrigin.getRate());
-        logger.info("PARAM -- Target Rate Limit: {}", rateLimiterTarget.getRate());
-
-        CqlTable cqlTableOrigin, cqlTableTarget = null;
-        this.originSession = new EnhancedSession(propertyHelper, originSession, true);
-        cqlTableOrigin = this.originSession.getCqlTable();
-        cqlTableOrigin.setFeatureMap(featureMap);
-
-        boolean allFeaturesValid = true;
-        if (targetSession != null) {
-            this.targetSession = new EnhancedSession(propertyHelper, targetSession, false);
-            cqlTableTarget = this.targetSession.getCqlTable();
-            cqlTableOrigin.setOtherCqlTable(cqlTableTarget);
-            cqlTableTarget.setOtherCqlTable(cqlTableOrigin);
-            cqlTableTarget.setFeatureMap(featureMap);
-            for (Feature f : featureMap.values()) {
-                if (!f.initializeAndValidate(cqlTableOrigin, cqlTableTarget)) {
-                    allFeaturesValid = false;
-                    logger.error("Feature {} is not valid.  Please check the configuration.", f.getClass().getName());
+        try {
+            rateLimiterOrigin = RateLimiter.create(propertyHelper.getInteger(KnownProperties.PERF_RATELIMIT_ORIGIN));
+            rateLimiterTarget = RateLimiter.create(propertyHelper.getInteger(KnownProperties.PERF_RATELIMIT_TARGET));
+
+            logger.info("PARAM -- Origin Rate Limit: {}", rateLimiterOrigin.getRate());
+            logger.info("PARAM -- Target Rate Limit: {}", rateLimiterTarget.getRate());
+
+            CqlTable cqlTableOrigin, cqlTableTarget = null;
+            this.originSession = new EnhancedSession(propertyHelper, originSession, true);
+            cqlTableOrigin = this.originSession.getCqlTable();
+            cqlTableOrigin.setFeatureMap(featureMap);
+
+            boolean allFeaturesValid = true;
+            if (targetSession != null) {
+                this.targetSession = new EnhancedSession(propertyHelper, targetSession, false);
+                cqlTableTarget = this.targetSession.getCqlTable();
+                cqlTableOrigin.setOtherCqlTable(cqlTableTarget);
+                cqlTableTarget.setOtherCqlTable(cqlTableOrigin);
+                cqlTableTarget.setFeatureMap(featureMap);
+                for (Feature f : featureMap.values()) {
+                    if (!f.initializeAndValidate(cqlTableOrigin, cqlTableTarget)) {
+                        allFeaturesValid = false;
+                        logger.error("Feature {} is not valid.  Please check the configuration.", f.getClass().getName());
+                    }
                 }
-            }
 
-            PKFactory pkFactory = new PKFactory(propertyHelper, cqlTableOrigin, cqlTableTarget);
-            this.originSession.setPKFactory(pkFactory);
-            this.targetSession.setPKFactory(pkFactory);
-        }
+                PKFactory pkFactory = new PKFactory(propertyHelper, cqlTableOrigin, cqlTableTarget);
+                this.originSession.setPKFactory(pkFactory);
+                this.targetSession.setPKFactory(pkFactory);
+            }
 
-        if (!allFeaturesValid) {
-            throw new RuntimeException("One or more features are not valid.  Please check the configuration.");
-        }
+            if (!allFeaturesValid) {
+                throw new RuntimeException("One or more features are not valid.  Please check the configuration.");
+            }
 
-        this.guardrailFeature = (Guardrail) cqlTableOrigin.getFeature(Featureset.GUARDRAIL_CHECK);
-        if (!guardrailFeature.initializeAndValidate(cqlTableOrigin, null)) {
-            allFeaturesValid = false;
-            logger.error("Feature {} is not valid.  Please check the configuration.",
-                    guardrailFeature.getClass().getName());
+            this.guardrailFeature = (Guardrail) cqlTableOrigin.getFeature(Featureset.GUARDRAIL_CHECK);
+            if (!guardrailFeature.initializeAndValidate(cqlTableOrigin, null)) {
+                allFeaturesValid = false;
+                logger.error("Feature {} is not valid.  Please check the configuration.",
+                        guardrailFeature.getClass().getName());
+            }
+        } catch (ClusterConfigurationException e) {
+            logger.error("Cluster configuration error may be present & detected: {}", e.getMessage());
+            logger.error("Please check your Cassandra cluster for token overlap issues. This usually happens when multiple nodes in the cluster were started simultaneously when the cluster was originally built.");
+            logger.error("You can verify this by running 'nodetool describering <keyspace>' and checking for overlapping token ranges.");
+            logger.error("In general, to fix token overlap in a cluster: 1) Rebuild the entire cluster by removing nodes 2) Re-add nodes one at a time 3) Run nodetool cleanup on each node 4) Verify with 'nodetool describering <keyspace>'");
+            throw e;
         }
     }
 
diff --git a/src/main/java/com/datastax/cdm/schema/ClusterConfigurationException.java b/src/main/java/com/datastax/cdm/schema/ClusterConfigurationException.java
@@ -0,0 +1,42 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.cdm.schema;
+
+/**
+ * Exception thrown when there are issues with the Cassandra cluster configuration,
+ * such as token overlap problems or other metadata-related issues.
+ */
+public class ClusterConfigurationException extends RuntimeException {
+    
+    /**
+     * Constructs a new exception with the specified detail message.
+     *
+     * @param message the detail message
+     */
+    public ClusterConfigurationException(String message) {
+        super(message);
+    }
+    
+    /**
+     * Constructs a new exception with the specified detail message and cause.
+     *
+     * @param message the detail message
+     * @param cause the cause of the exception
+     */
+    public ClusterConfigurationException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}
diff --git a/src/main/java/com/datastax/cdm/schema/CqlTable.java b/src/main/java/com/datastax/cdm/schema/CqlTable.java
@@ -15,8 +15,10 @@
  */
 package com.datastax.cdm.schema;
 
+import java.math.BigInteger;
 import java.time.Instant;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -42,7 +44,9 @@
 import com.datastax.oss.driver.api.core.ConsistencyLevel;
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.cql.ResultSet;
 import com.datastax.oss.driver.api.core.cql.Row;
+import com.datastax.oss.driver.api.core.metadata.TokenMap;
 import com.datastax.oss.driver.api.core.metadata.Metadata;
 import com.datastax.oss.driver.api.core.metadata.schema.ColumnMetadata;
 import com.datastax.oss.driver.api.core.metadata.schema.KeyspaceMetadata;
@@ -431,12 +435,33 @@ protected Metadata fetchMetadataFromSession(CqlSession cqlSession) {
 
     private void setCqlMetadata(CqlSession cqlSession) {
         Metadata metadata = fetchMetadataFromSession(cqlSession);
-
-        String partitionerName = metadata.getTokenMap().get().getPartitionerName();
-        if (null != partitionerName && partitionerName.endsWith("RandomPartitioner"))
-            this.hasRandomPartitioner = true;
-        else
-            this.hasRandomPartitioner = false;
+        
+        // Check for token overlap in the specific keyspace
+        if (hasTokenOverlap(cqlSession, this.keyspaceName)) {
+            throw new ClusterConfigurationException(
+                "Token overlap detected in keyspace '" + this.keyspaceName + "'. This usually happens when multiple nodes " +
+                "were started simultaneously. To fix: 1) Restart nodes one at a time 2) Run 'nodetool cleanup' " +
+                "on each node 3) Verify with 'nodetool describering " + this.keyspaceName + "'.");
+        }
+        
+        // Add proper Optional handling for token map
+        Optional<TokenMap> tokenMapOpt = metadata.getTokenMap();
+        if (!tokenMapOpt.isPresent()) {
+            throw new ClusterConfigurationException(
+                "Token map is not available. This could indicate a cluster configuration issue.");
+        }
+        
+        try {
+            String partitionerName = tokenMapOpt.get().getPartitionerName();
+            if (null != partitionerName && partitionerName.endsWith("RandomPartitioner"))
+                this.hasRandomPartitioner = true;
+            else
+                this.hasRandomPartitioner = false;
+        } catch (Exception e) {
+            throw new ClusterConfigurationException(
+                "Error accessing token map: " + e.getMessage() + 
+                ". This may indicate token overlap in the Cassandra cluster. Check your cluster configuration.", e);
+        }
 
         Optional<KeyspaceMetadata> keyspaceMetadataOpt = metadata.getKeyspace(formatName(this.keyspaceName));
         if (!keyspaceMetadataOpt.isPresent()) {
@@ -563,5 +588,81 @@ protected static ConsistencyLevel mapToConsistencyLevel(String level) {
 
         return retVal;
     }
-
+    
+    /**
+     * Checks if the specified keyspace has token overlap issues by querying the system.size_estimates table.
+     * 
+     * @param cqlSession The CQL session to use for executing the query
+     * @param keyspaceName The name of the keyspace to check
+     * @return true if token overlap is detected, false otherwise
+     */
+    private boolean hasTokenOverlap(CqlSession cqlSession, String keyspaceName) {
+        try {
+            // Execute query to check for token ranges for the specific keyspace
+            String query = "SELECT start_token, end_token FROM system.size_estimates WHERE keyspace_name = ?";
+            ResultSet rs = cqlSession.execute(query, keyspaceName);
+            
+            // Create a list to store token ranges for the keyspace
+            List<TokenRange> ranges = new ArrayList<>();
+            
+            // Process the results
+            for (Row row : rs) {
+                BigInteger startToken = new BigInteger(row.getString("start_token"));
+                BigInteger endToken = new BigInteger(row.getString("end_token"));
+                ranges.add(new TokenRange(startToken, endToken));
+            }
+            
+            // Check for overlaps
+            if (hasOverlappingTokens(ranges)) {
+                logger.error("Token overlap detected in keyspace: {}", keyspaceName);
+                return true;
+            }
+            
+            return false;
+        } catch (Exception e) {
+            logger.warn("Could not check for token overlap in keyspace {}: {}", keyspaceName, e.getMessage());
+            return false;
+        }
+    }
+    
+    /**
+     * Determines if there are overlapping token ranges in the provided list.
+     * 
+     * @param ranges List of token ranges to check
+     * @return true if any ranges overlap, false otherwise
+     */
+    private boolean hasOverlappingTokens(List<TokenRange> ranges) {
+        // Sort ranges by start token
+        Collections.sort(ranges);
+        
+        // Check for overlaps
+        for (int i = 0; i < ranges.size() - 1; i++) {
+            TokenRange current = ranges.get(i);
+            TokenRange next = ranges.get(i + 1);
+            
+            if (current.endToken.compareTo(next.startToken) > 0) {
+                return true;
+            }
+        }
+        
+        return false;
+    }
+    
+    /**
+     * Helper class to represent a token range with Comparable implementation for sorting.
+     */
+    private static class TokenRange implements Comparable<TokenRange> {
+        BigInteger startToken;
+        BigInteger endToken;
+        
+        TokenRange(BigInteger startToken, BigInteger endToken) {
+            this.startToken = startToken;
+            this.endToken = endToken;
+        }
+        
+        @Override
+        public int compareTo(TokenRange other) {
+            return this.startToken.compareTo(other.startToken);
+        }
+    }
 }
