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

Merge branch 'cassandra-4.1' into trunk

This commit is contained in:
Paulo Motta 2023-03-24 12:17:44 -04:00
parent 2e2a49fcdc 7c2f97cd29
commit ffe669106b
8 changed files with 88 additions and 181 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -124,6 +124,7 @@
* Rename DisableFlag class to EnableFlag on guardrails (CASSANDRA-17544)
4.1.2
* Do not submit hints when hinted_handoff_enabled=false (CASSANDRA-18304)
* Fix COPY ... TO STDOUT behavior in cqlsh (CASSANDRA-18353)
* Remove six and Py2SaferScanner merge cruft (CASSANDRA-18354)

2
src/java/org/apache/cassandra/batchlog/BatchlogManager.java
View File

@ -505,7 +505,7 @@ public class BatchlogManager implements BatchlogManagerMBean
ReplayWriteResponseHandler<Mutation> handler = new ReplayWriteResponseHandler<>(replicaPlan, mutation, nanoTime());
Message<Mutation> message = Message.outWithFlag(MUTATION_REQ, mutation, MessageFlag.CALL_BACK_ON_FAILURE);
for (Replica replica : liveRemoteOnly.all())
MessagingService.instance().sendWriteWithCallback(message, replica, handler, false);
MessagingService.instance().sendWriteWithCallback(message, replica, handler);
return handler;
}

4
src/java/org/apache/cassandra/net/MessagingService.java
View File

@ -376,10 +376,10 @@ public class MessagingService extends MessagingServiceMBeanImpl
* @param handler callback interface which is used to pass the responses or
* suggest that a timeout occurred to the invoker of the send().
*/
public void sendWriteWithCallback(Message message, Replica to, AbstractWriteResponseHandler<?> handler, boolean allowHints)
public void sendWriteWithCallback(Message message, Replica to, AbstractWriteResponseHandler<?> handler)
{
assert message.callBackOnFailure();
callbacks.addWithExpiration(handler, message, to, handler.consistencyLevel(), allowHints);
callbacks.addWithExpiration(handler, message, to);
send(message, to.endpoint(), null);
}

69
src/java/org/apache/cassandra/net/RequestCallbacks.java
View File

@ -31,17 +31,12 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.exceptions.RequestFailureReason;
import org.apache.cassandra.io.IVersionedAsymmetricSerializer;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.metrics.InternodeOutboundMetrics;
import org.apache.cassandra.service.AbstractWriteResponseHandler;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.paxos.Commit;
import org.apache.cassandra.utils.FBUtilities;
import static java.lang.String.format;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
@ -99,22 +94,18 @@ public class RequestCallbacks implements OutboundMessageCallbacks
/**
* Register the provided {@link RequestCallback}, inferring expiry and id from the provided {@link Message}.
*/
void addWithExpiration(RequestCallback cb, Message message, InetAddressAndPort to)
public void addWithExpiration(RequestCallback<?> cb, Message<?> message, InetAddressAndPort to)
{
// mutations need to call the overload with a ConsistencyLevel
// mutations need to call the overload
assert message.verb() != Verb.MUTATION_REQ && message.verb() != Verb.COUNTER_MUTATION_REQ;
CallbackInfo previous = callbacks.put(key(message.id(), to), new CallbackInfo(message, to, cb));
assert previous == null : format("Callback already exists for id %d/%s! (%s)", message.id(), to, previous);
}
public void addWithExpiration(AbstractWriteResponseHandler<?> cb,
Message<?> message,
Replica to,
ConsistencyLevel consistencyLevel,
boolean allowHints)
public void addWithExpiration(AbstractWriteResponseHandler<?> cb, Message<?> message, Replica to)
{
assert message.verb() == Verb.MUTATION_REQ || message.verb() == Verb.COUNTER_MUTATION_REQ || message.verb() == Verb.PAXOS_COMMIT_REQ;
CallbackInfo previous = callbacks.put(key(message.id(), to.endpoint()), new WriteCallbackInfo(message, to, cb, consistencyLevel, allowHints));
CallbackInfo previous = callbacks.put(key(message.id(), to.endpoint()), new CallbackInfo(message, to.endpoint(), cb));
assert previous == null : format("Callback already exists for id %d/%s! (%s)", message.id(), to.endpoint(), previous);
}
@ -274,11 +265,6 @@ public class RequestCallbacks implements OutboundMessageCallbacks
return atNano > expiresAtNanos;
}
boolean shouldHint()
{
return false;
}
boolean invokeOnFailure()
{
return callback.invokeOnFailure();
@ -290,53 +276,6 @@ public class RequestCallbacks implements OutboundMessageCallbacks
}
}
// FIXME: shouldn't need a specialized container for write callbacks; hinting should be part of
// AbstractWriteResponseHandler implementation.
static class WriteCallbackInfo extends CallbackInfo
{
// either a Mutation, or a Paxos Commit (MessageOut)
private final Object mutation;
private final Replica replica;
@VisibleForTesting
WriteCallbackInfo(Message message, Replica replica, RequestCallback<?> callback, ConsistencyLevel consistencyLevel, boolean allowHints)
{
super(message, replica.endpoint(), callback);
this.mutation = shouldHint(allowHints, message, consistencyLevel) ? message.payload : null;
//Local writes shouldn't go through messaging service (https://issues.apache.org/jira/browse/CASSANDRA-10477)
//noinspection AssertWithSideEffects
assert !peer.equals(FBUtilities.getBroadcastAddressAndPort());
this.replica = replica;
}
public boolean shouldHint()
{
return mutation != null && StorageProxy.shouldHint(replica);
}
public Replica getReplica()
{
return replica;
}
public Mutation mutation()
{
return getMutation(mutation);
}
private static Mutation getMutation(Object object)
{
assert object instanceof Commit || object instanceof Mutation : object;
return object instanceof Commit ? ((Commit) object).makeMutation()
: (Mutation) object;
}
private static boolean shouldHint(boolean allowHints, Message sentMessage, ConsistencyLevel consistencyLevel)
{
return allowHints && sentMessage.verb() != Verb.COUNTER_MUTATION_REQ && consistencyLevel != ConsistencyLevel.ANY;
}
}
@Override
public void onOverloaded(Message<?> message, InetAddressAndPort peer)
{

2
src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java
View File

@ -292,7 +292,7 @@ public abstract class AbstractWriteResponseHandler<T> implements RequestCallback
if (blockFor() + n > candidateReplicaCount())
signal();
if (hintOnFailure != null)
if (hintOnFailure != null && StorageProxy.shouldHint(replicaPlan.lookup(from)))
StorageProxy.submitHint(hintOnFailure.get(), replicaPlan.lookup(from), null);
}

10
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -795,7 +795,7 @@ public class StorageProxy implements StorageProxyMBean
if (replica.isSelf())
commitPaxosLocal(replica, message, responseHandler);
else
MessagingService.instance().sendWriteWithCallback(message, replica, responseHandler, allowHints && shouldHint(replica));
MessagingService.instance().sendWriteWithCallback(message, replica, responseHandler);
}
else
{
@ -1560,7 +1560,7 @@ public class StorageProxy implements StorageProxyMBean
if (localDc != null)
{
for (Replica destination : localDc)
MessagingService.instance().sendWriteWithCallback(message, destination, responseHandler, true);
MessagingService.instance().sendWriteWithCallback(message, destination, responseHandler);
}
if (dcGroups != null)
{
@ -1602,7 +1602,7 @@ public class StorageProxy implements StorageProxyMBean
for (Replica replica : forwardToReplicas)
{
MessagingService.instance().callbacks.addWithExpiration(handler, message, replica, handler.replicaPlan.consistencyLevel(), true);
MessagingService.instance().callbacks.addWithExpiration(handler, message, replica);
logger.trace("Adding FWD message to {}@{}", message.id(), replica);
}
@ -1617,7 +1617,7 @@ public class StorageProxy implements StorageProxyMBean
target = targets.get(0);
}
MessagingService.instance().sendWriteWithCallback(message, target, handler, true);
MessagingService.instance().sendWriteWithCallback(message, target, handler);
logger.trace("Sending message to {}@{}", message.id(), target);
}
@ -1729,7 +1729,7 @@ public class StorageProxy implements StorageProxyMBean
Tracing.trace("Enqueuing counter update to {}", replica);
Message message = Message.outWithFlag(Verb.COUNTER_MUTATION_REQ, cm, MessageFlag.CALL_BACK_ON_FAILURE);
MessagingService.instance().sendWriteWithCallback(message, replica, responseHandler, false);
MessagingService.instance().sendWriteWithCallback(message, replica, responseHandler);
return responseHandler;
}
}

74
test/distributed/org/apache/cassandra/distributed/test/HintsDisabledTest.java Normal file
View File

@ -0,0 +1,74 @@
/*
* 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.distributed.test;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import com.google.monitoring.runtime.instrumentation.common.util.concurrent.Uninterruptibles;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.IMessageFilters;
import org.apache.cassandra.metrics.StorageMetrics;
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
import static org.apache.cassandra.net.Verb.MUTATION_REQ;
import static org.assertj.core.api.Assertions.assertThat;
public class HintsDisabledTest extends TestBaseImpl
{
@Test
public void testHintedHandoffDisabled() throws IOException
{
try (Cluster cluster = init(Cluster.build(2)
.withDataDirCount(1)
.withConfig(config -> config.with(NETWORK, GOSSIP)
.set("write_request_timeout", "10ms")
.set("hinted_handoff_enabled", false))
.start(), 2))
{
String createTableStatement = String.format("CREATE TABLE %s.cf (k text PRIMARY KEY, c1 text) " +
"WITH compaction = {'class': 'SizeTieredCompactionStrategy', 'enabled': 'false'} ", KEYSPACE);
cluster.schemaChange(createTableStatement);
// Drop all messages from node1 to node2 so hints should be created
IMessageFilters.Filter drop1to2 = cluster.filters().verbs(MUTATION_REQ.id).from(1).to(2).drop();
cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.cf (k, c1) VALUES (?, ?) USING TIMESTAMP 1;"),
ConsistencyLevel.ONE,
String.valueOf(1),
String.valueOf(1));
// Wait 15ms for write to timeout (write_request_timeout=10ms)
Uninterruptibles.sleepUninterruptibly(15, TimeUnit.MILLISECONDS);
// Check that no hints were created on node1
assertThat(cluster.get(1).callOnInstance(() -> Long.valueOf(StorageMetrics.totalHints.getCount()))).isEqualTo(0L);
}
}
private static int getNumberOfSSTables(Cluster cluster, int node)
{
return cluster.get(node).callOnInstance(() -> ColumnFamilyStore.getIfExists(KEYSPACE, "cf").getLiveSSTables().size());
}
}

107
test/unit/org/apache/cassandra/net/WriteCallbackInfoTest.java
View File

@ -1,107 +0,0 @@
/*
* 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.net;
import java.util.UUID;
import org.junit.After;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.Assert;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.RegularAndStaticColumns;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.schema.MockSchema;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.paxos.Commit;
import org.apache.cassandra.utils.ByteBufferUtil;
import static org.apache.cassandra.locator.ReplicaUtils.full;
import static org.apache.cassandra.service.paxos.Ballot.Flag.NONE;
import static org.apache.cassandra.service.paxos.BallotGenerator.Global.nextBallot;
public class WriteCallbackInfoTest
{
private InetAddressAndPort testEp;
@BeforeClass
public static void initDD()
{
DatabaseDescriptor.daemonInitialization();
}
@Before
public void setup() throws Exception
{
testEp = InetAddressAndPort.getByName("192.168.1.1");
StorageService.instance.getTokenMetadata().updateHostId(UUID.randomUUID(), testEp);
}
@After
public void teardown()
{
StorageService.instance.getTokenMetadata().removeEndpoint(testEp);
}
@Test
public void testShouldHint() throws Exception
{
testShouldHint(Verb.COUNTER_MUTATION_REQ, ConsistencyLevel.ALL, true, false);
for (Verb verb : new Verb[] { Verb.PAXOS_COMMIT_REQ, Verb.MUTATION_REQ })
{
testShouldHint(verb, ConsistencyLevel.ALL, true, true);
testShouldHint(verb, ConsistencyLevel.ANY, true, false);
testShouldHint(verb, ConsistencyLevel.ALL, false, false);
}
}
private void testShouldHint(Verb verb, ConsistencyLevel cl, boolean allowHints, boolean expectHint)
{
TableMetadata metadata = MockSchema.newTableMetadata("", "");
Object payload = verb == Verb.PAXOS_COMMIT_REQ
? new Commit(nextBallot(NONE), new PartitionUpdate.Builder(metadata, ByteBufferUtil.EMPTY_BYTE_BUFFER, RegularAndStaticColumns.NONE, 1).build())
: new Mutation(PartitionUpdate.simpleBuilder(metadata, "").build());
RequestCallbacks.WriteCallbackInfo wcbi = new RequestCallbacks.WriteCallbackInfo(Message.out(verb, payload), full(testEp), null, cl, allowHints);
Assert.assertEquals(expectHint, wcbi.shouldHint());
if (expectHint)
{
Assert.assertNotNull(wcbi.mutation());
}
else
{
boolean fail = false;
try
{
wcbi.mutation();
}
catch (Throwable t)
{
fail = true;
}
Assert.assertTrue(fail);
}
}
}