natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make deprecated repair methods backward-compatible with previous notification service 

patch by Paulo Motta; reviewed by Yuki Morishita for CASSANDRA-11430
This commit is contained in:
Paulo Motta 2016-04-05 16:58:06 -03:00 committed by Yuki Morishita
parent e22faeb8c5
commit 3557d2e05c
6 changed files with 270 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

10
src/java/org/apache/cassandra/repair/RepairRunnable.java
View File

@ -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);

11
src/java/org/apache/cassandra/service/ActiveRepairService.java
View File

@ -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);

28
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -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);
}

108
src/java/org/apache/cassandra/utils/progress/jmx/LegacyJMXProgressSupport.java Normal file
View File

@ -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]);
}
}

118
test/unit/org/apache/cassandra/utils/progress/jmx/LegacyJMXProgressSupportTest.java Normal file
View File

@ -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());
}
}