cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From yu...@apache.org
Subject [1/6] cassandra git commit: Make deprecated repair methods backward-compatible with previous notification service
Date Mon, 11 Apr 2016 16:42:18 GMT
Repository: cassandra
Updated Branches:
  refs/heads/cassandra-2.2 e22faeb8c -> 3557d2e05
  refs/heads/cassandra-3.0 063b37614 -> f0cd3261b
  refs/heads/trunk 2ae587f5c -> 1aeeff47a


Make deprecated repair methods backward-compatible with previous notification service

patch by Paulo Motta; reviewed by Yuki Morishita for CASSANDRA-11430


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/3557d2e0
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/3557d2e0
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/3557d2e0

Branch: refs/heads/cassandra-2.2
Commit: 3557d2e05c8d1059562de2a91c1b33b4fcfcc6eb
Parents: e22faeb
Author: Paulo Motta <pauloricardomg@gmail.com>
Authored: Tue Apr 5 16:58:06 2016 -0300
Committer: Yuki Morishita <yukim@apache.org>
Committed: Mon Apr 11 11:28:44 2016 -0500

----------------------------------------------------------------------
 CHANGES.txt                                     |   1 +
 .../apache/cassandra/repair/RepairRunnable.java |  10 ++
 .../cassandra/service/ActiveRepairService.java  |  11 ++
 .../cassandra/service/StorageService.java       |  28 ++++-
 .../progress/jmx/LegacyJMXProgressSupport.java  | 108 +++++++++++++++++
 .../jmx/LegacyJMXProgressSupportTest.java       | 118 +++++++++++++++++++
 6 files changed, 270 insertions(+), 6 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/3557d2e0/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index b6438b8..e935e57 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 2.2.6
+ * Make deprecated repair methods backward-compatible with previous notification service
(CASSANDRA-11430)
  * IncomingStreamingConnection version check message wrong (CASSANDRA-11462)
  * DatabaseDescriptor should log stacktrace in case of Eception during seed provider creation
(CASSANDRA-11312)
  * Use canonical path for directory in SSTable descriptor (CASSANDRA-10587)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/3557d2e0/src/java/org/apache/cassandra/repair/RepairRunnable.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/repair/RepairRunnable.java b/src/java/org/apache/cassandra/repair/RepairRunnable.java
index 91ac82a..d2b6ab6 100644
--- a/src/java/org/apache/cassandra/repair/RepairRunnable.java
+++ b/src/java/org/apache/cassandra/repair/RepairRunnable.java
@@ -227,6 +227,11 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
             {
                 public void onSuccess(RepairSessionResult result)
                 {
+                    /**
+                     * If the success message below is modified, it must also be updated
on
+                     * {@link org.apache.cassandra.utils.progress.jmx.LegacyJMXProgressSupport}
+                     * for backward-compatibility support.
+                     */
                     String message = String.format("Repair session %s for range %s finished",
session.getId(),
                                                    session.getRange().toString());
                     logger.info(message);
@@ -238,6 +243,11 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
 
                 public void onFailure(Throwable t)
                 {
+                    /**
+                     * If the failure message below is modified, it must also be updated
on
+                     * {@link org.apache.cassandra.utils.progress.jmx.LegacyJMXProgressSupport}
+                     * for backward-compatibility support.
+                     */
                     String message = String.format("Repair session %s for range %s failed
with error %s",
                                                    session.getId(), session.getRange().toString(),
t.getMessage());
                     logger.error(message, t);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/3557d2e0/src/java/org/apache/cassandra/service/ActiveRepairService.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/ActiveRepairService.java b/src/java/org/apache/cassandra/service/ActiveRepairService.java
index 39be051..21cdeae 100644
--- a/src/java/org/apache/cassandra/service/ActiveRepairService.java
+++ b/src/java/org/apache/cassandra/service/ActiveRepairService.java
@@ -77,6 +77,17 @@ import org.apache.cassandra.utils.concurrent.Refs;
  */
 public class ActiveRepairService
 {
+    /**
+     * @deprecated this statuses are from the previous JMX notification service,
+     * which will be deprecated on 4.0. For statuses of the new notification
+     * service, see {@link org.apache.cassandra.streaming.StreamEvent.ProgressEvent}
+     */
+    @Deprecated
+    public static enum Status
+    {
+        STARTED, SESSION_SUCCESS, SESSION_FAILED, FINISHED
+    }
+
     public static CassandraVersion SUPPORTS_GLOBAL_PREPARE_FLAG_VERSION = new CassandraVersion("2.2.1");
 
     private static final Logger logger = LoggerFactory.getLogger(ActiveRepairService.class);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/3557d2e0/src/java/org/apache/cassandra/service/StorageService.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index bca5996..dedc823 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -171,6 +171,7 @@ import org.apache.cassandra.utils.WrappedRunnable;
 import org.apache.cassandra.utils.progress.ProgressEvent;
 import org.apache.cassandra.utils.progress.ProgressEventType;
 import org.apache.cassandra.utils.progress.jmx.JMXProgressSupport;
+import org.apache.cassandra.utils.progress.jmx.LegacyJMXProgressSupport;
 
 /**
  * This abstraction contains the token/identifier of this node
@@ -186,6 +187,13 @@ public class StorageService extends NotificationBroadcasterSupport implements
IE
 
     private final JMXProgressSupport progressSupport = new JMXProgressSupport(this);
 
+    /**
+     * @deprecated backward support to previous notification interface
+     * Will be removed on 4.0
+     */
+    @Deprecated
+    private final LegacyJMXProgressSupport legacyProgressSupport;
+
     private static int getRingDelay()
     {
         String newdelay = System.getProperty("cassandra.ring_delay_ms");
@@ -309,6 +317,8 @@ public class StorageService extends NotificationBroadcasterSupport implements
IE
             throw new RuntimeException(e);
         }
 
+        legacyProgressSupport = new LegacyJMXProgressSupport(this, jmxObjectName);
+
         /* register the verb handlers */
         MessagingService.instance().registerVerbHandlers(MessagingService.Verb.MUTATION,
new MutationVerbHandler());
         MessagingService.instance().registerVerbHandlers(MessagingService.Verb.READ_REPAIR,
new ReadRepairVerbHandler());
@@ -2896,7 +2906,7 @@ public class StorageService extends NotificationBroadcasterSupport implements
IE
                 option.getRanges().addAll(getLocalRanges(keyspace));
             }
         }
-        return forceRepairAsync(keyspace, option);
+        return forceRepairAsync(keyspace, option, false);
     }
 
     @Deprecated
@@ -2962,9 +2972,10 @@ public class StorageService extends NotificationBroadcasterSupport
implements IE
                 options.getColumnFamilies().add(columnFamily);
             }
         }
-        return forceRepairAsync(keyspace, options);
+        return forceRepairAsync(keyspace, options, true);
     }
 
+    @Deprecated
     public int forceRepairAsync(String keyspace,
                                 boolean isSequential,
                                 boolean isLocal,
@@ -2980,6 +2991,7 @@ public class StorageService extends NotificationBroadcasterSupport implements
IE
         return forceRepairAsync(keyspace, isSequential, dataCenters, null, primaryRange,
fullRepair, columnFamilies);
     }
 
+    @Deprecated
     public int forceRepairRangeAsync(String beginToken,
                                      String endToken,
                                      String keyspaceName,
@@ -2994,6 +3006,7 @@ public class StorageService extends NotificationBroadcasterSupport implements
IE
                                      dataCenters, hosts, fullRepair, columnFamilies);
     }
 
+    @Deprecated
     public int forceRepairRangeAsync(String beginToken,
                                      String endToken,
                                      String keyspaceName,
@@ -3036,9 +3049,10 @@ public class StorageService extends NotificationBroadcasterSupport
implements IE
 
         logger.info("starting user-requested repair of range {} for keyspace {} and column
families {}",
                     repairingRange, keyspaceName, columnFamilies);
-        return forceRepairAsync(keyspaceName, options);
+        return forceRepairAsync(keyspaceName, options, true);
     }
 
+    @Deprecated
     public int forceRepairRangeAsync(String beginToken,
                                      String endToken,
                                      String keyspaceName,
@@ -3093,17 +3107,17 @@ public class StorageService extends NotificationBroadcasterSupport
implements IE
         return repairingRange;
     }
 
-    public int forceRepairAsync(String keyspace, RepairOption options)
+    public int forceRepairAsync(String keyspace, RepairOption options, boolean legacy)
     {
         if (options.getRanges().isEmpty() || Keyspace.open(keyspace).getReplicationStrategy().getReplicationFactor()
< 2)
             return 0;
 
         int cmd = nextRepairCommand.incrementAndGet();
-        new Thread(createRepairTask(cmd, keyspace, options)).start();
+        new Thread(createRepairTask(cmd, keyspace, options, legacy)).start();
         return cmd;
     }
 
-    private FutureTask<Object> createRepairTask(final int cmd, final String keyspace,
final RepairOption options)
+    private FutureTask<Object> createRepairTask(final int cmd, final String keyspace,
final RepairOption options, boolean legacy)
     {
         if (!options.getDataCenters().isEmpty() && !options.getDataCenters().contains(DatabaseDescriptor.getLocalDataCenter()))
         {
@@ -3112,6 +3126,8 @@ public class StorageService extends NotificationBroadcasterSupport implements
IE
 
         RepairRunnable task = new RepairRunnable(this, cmd, options, keyspace);
         task.addProgressListener(progressSupport);
+        if (legacy)
+            task.addProgressListener(legacyProgressSupport);
         return new FutureTask<>(task, null);
     }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/3557d2e0/src/java/org/apache/cassandra/utils/progress/jmx/LegacyJMXProgressSupport.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/progress/jmx/LegacyJMXProgressSupport.java
b/src/java/org/apache/cassandra/utils/progress/jmx/LegacyJMXProgressSupport.java
new file mode 100644
index 0000000..fae6f2a
--- /dev/null
+++ b/src/java/org/apache/cassandra/utils/progress/jmx/LegacyJMXProgressSupport.java
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.cassandra.utils.progress.jmx;
+
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.regex.Pattern;
+import javax.management.Notification;
+import javax.management.NotificationBroadcasterSupport;
+import javax.management.ObjectName;
+
+import com.google.common.base.Optional;
+
+import org.apache.cassandra.utils.progress.ProgressEvent;
+import org.apache.cassandra.utils.progress.ProgressListener;
+
+import static org.apache.cassandra.service.ActiveRepairService.Status;
+
+/**
+ * ProgressListener that translates ProgressEvent to legacy JMX Notification message (backward
compatibility support)
+ */
+public class LegacyJMXProgressSupport implements ProgressListener
+{
+    protected static final Pattern SESSION_FAILED_MATCHER = Pattern.compile("Repair session
.* for range .* failed with error .*");
+    protected static final Pattern SESSION_SUCCESS_MATCHER = Pattern.compile("Repair session
.* for range .* finished");
+
+    private final AtomicLong notificationSerialNumber = new AtomicLong();
+    private final ObjectName jmxObjectName;
+
+    private final NotificationBroadcasterSupport broadcaster;
+
+    public LegacyJMXProgressSupport(NotificationBroadcasterSupport broadcaster,
+                                    ObjectName jmxObjectName)
+    {
+        this.broadcaster = broadcaster;
+        this.jmxObjectName = jmxObjectName;
+    }
+
+    @Override
+    public void progress(String tag, ProgressEvent event)
+    {
+        if (tag.startsWith("repair:"))
+        {
+            Optional<int[]> legacyUserData = getLegacyUserdata(tag, event);
+            if (legacyUserData.isPresent())
+            {
+                Notification jmxNotification = new Notification("repair", jmxObjectName,
notificationSerialNumber.incrementAndGet(), event.getMessage());
+                jmxNotification.setUserData(legacyUserData.get());
+                broadcaster.sendNotification(jmxNotification);
+            }
+        }
+    }
+
+    protected static Optional<int[]> getLegacyUserdata(String tag, ProgressEvent event)
+    {
+        Optional<Status> status = getStatus(event);
+        if (status.isPresent())
+        {
+            int[] result = new int[2];
+            result[0] = getCmd(tag);
+            result[1] = status.get().ordinal();
+            return Optional.of(result);
+        }
+        return Optional.absent();
+    }
+
+    protected static Optional<Status> getStatus(ProgressEvent event)
+    {
+        switch (event.getType())
+        {
+            case START:
+                return Optional.of(Status.STARTED);
+            case COMPLETE:
+                return Optional.of(Status.FINISHED);
+            case PROGRESS:
+                if (SESSION_FAILED_MATCHER.matcher(event.getMessage()).matches())
+                {
+                    return Optional.of(Status.SESSION_FAILED);
+                }
+                else if (SESSION_SUCCESS_MATCHER.matcher(event.getMessage()).matches())
+                {
+                    return Optional.of(Status.SESSION_SUCCESS);
+                }
+        }
+
+        return Optional.absent();
+    }
+
+    protected static int getCmd(String tag)
+    {
+        return Integer.valueOf(tag.split(":")[1]);
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/3557d2e0/test/unit/org/apache/cassandra/utils/progress/jmx/LegacyJMXProgressSupportTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/utils/progress/jmx/LegacyJMXProgressSupportTest.java
b/test/unit/org/apache/cassandra/utils/progress/jmx/LegacyJMXProgressSupportTest.java
new file mode 100644
index 0000000..efa4a27
--- /dev/null
+++ b/test/unit/org/apache/cassandra/utils/progress/jmx/LegacyJMXProgressSupportTest.java
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.cassandra.utils.progress.jmx;
+
+import java.util.UUID;
+
+import com.google.common.base.Optional;
+import org.junit.Test;
+
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.service.ActiveRepairService;
+import org.apache.cassandra.utils.progress.ProgressEvent;
+import org.apache.cassandra.utils.progress.ProgressEventType;
+
+import static org.junit.Assert.*;
+
+
+public class LegacyJMXProgressSupportTest
+{
+
+    @Test
+    public void testSessionSuccess()
+    {
+        int cmd = 321;
+        String message = String.format("Repair session %s for range %s finished", UUID.randomUUID(),
+                                       new Range<Token>(new Murmur3Partitioner.LongToken(3),
new Murmur3Partitioner.LongToken(4)));
+        Optional<int[]> result = LegacyJMXProgressSupport.getLegacyUserdata(String.format("repair:%d",
cmd),
+                                                            new ProgressEvent(ProgressEventType.PROGRESS,
2, 10, message));
+        assertTrue(result.isPresent());
+        assertArrayEquals(new int[]{ cmd, ActiveRepairService.Status.SESSION_SUCCESS.ordinal()
}, result.get());
+    }
+
+    @Test
+    public void testSessionFailed()
+    {
+        int cmd = 321;
+        String message = String.format("Repair session %s for range %s failed with error
%s", UUID.randomUUID(),
+                                       new Range<Token>(new Murmur3Partitioner.LongToken(3),
new Murmur3Partitioner.LongToken(4)).toString(),
+                                       new RuntimeException("error"));
+        Optional<int[]> result = LegacyJMXProgressSupport.getLegacyUserdata(String.format("repair:%d",
cmd),
+                                                                            new ProgressEvent(ProgressEventType.PROGRESS,
2, 10, message));
+        assertTrue(result.isPresent());
+        assertArrayEquals(new int[]{ cmd, ActiveRepairService.Status.SESSION_FAILED.ordinal()
}, result.get());
+    }
+
+    @Test
+    public void testStarted()
+    {
+        int cmd = 321;
+        Optional<int[]> result = LegacyJMXProgressSupport.getLegacyUserdata(String.format("repair:%d",
cmd),
+                                                                            new ProgressEvent(ProgressEventType.START,
+                                                                                        
     0, 100, "bla"));
+        assertTrue(result.isPresent());
+        assertArrayEquals(new int[]{ cmd, ActiveRepairService.Status.STARTED.ordinal() },
result.get());
+    }
+
+    @Test
+    public void testFinished()
+    {
+        int cmd = 321;
+        Optional<int[]> result = LegacyJMXProgressSupport.getLegacyUserdata(String.format("repair:%d",
cmd),
+                                                                         new ProgressEvent(ProgressEventType.COMPLETE,
+                                                                                        
  2, 10, "bla"));
+        assertTrue(result.isPresent());
+        assertArrayEquals(new int[]{ cmd, ActiveRepairService.Status.FINISHED.ordinal() },
result.get());
+    }
+
+    /*
+    States not mapped to the legacy notification
+     */
+    @Test
+    public void testNone()
+    {
+        int cmd = 33;
+        Optional<int[]> result = LegacyJMXProgressSupport.getLegacyUserdata(String.format("repair:%d",
cmd),
+                                                                         new ProgressEvent(ProgressEventType.ERROR,
2, 10, "bla"));
+        assertFalse(result.isPresent());
+
+        cmd = 33;
+        result = LegacyJMXProgressSupport.getLegacyUserdata(String.format("repair:%d", cmd),
+                                                                         new ProgressEvent(ProgressEventType.SUCCESS,
2, 10, "bla"));
+        assertFalse(result.isPresent());
+
+        cmd = 43;
+        result = LegacyJMXProgressSupport.getLegacyUserdata(String.format("repair:%d", cmd),
+                                                            new ProgressEvent(ProgressEventType.PROGRESS,
2, 10, "bla"));
+        assertFalse(result.isPresent());
+
+        cmd = 1;
+        result = LegacyJMXProgressSupport.getLegacyUserdata(String.format("repair:%d", cmd),
+                                                            new ProgressEvent(ProgressEventType.ABORT,
2, 10, "bla"));
+        assertFalse(result.isPresent());
+
+        cmd = 9;
+        result = LegacyJMXProgressSupport.getLegacyUserdata(String.format("repair:%d", cmd),
+                                                            new ProgressEvent(ProgressEventType.NOTIFICATION,
2, 10, "bla"));
+        assertFalse(result.isPresent());
+    }
+
+}


Mime
View raw message