diff --git a/src/java/org/apache/cassandra/net/AbstractMessageHandler.java b/src/java/org/apache/cassandra/net/AbstractMessageHandler.java index 0ef03990fd70..5f190727dd42 100644 --- a/src/java/org/apache/cassandra/net/AbstractMessageHandler.java +++ b/src/java/org/apache/cassandra/net/AbstractMessageHandler.java @@ -31,7 +31,6 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.concurrent.ManyToOneConcurrentLinkedQueue; -import org.apache.cassandra.metrics.ClientMetrics; import org.apache.cassandra.net.FrameDecoder.CorruptFrame; import org.apache.cassandra.net.FrameDecoder.Frame; import org.apache.cassandra.net.FrameDecoder.FrameProcessor; @@ -292,7 +291,8 @@ private void onGlobalReserveCapacityRegained(Limit globalReserve, long elapsedNa onReserveCapacityRegained(endpointReserveCapacity, globalReserve, elapsedNanos); } - private void onReserveCapacityRegained(Limit endpointReserve, Limit globalReserve, long elapsedNanos) + @VisibleForTesting + void onReserveCapacityRegained(Limit endpointReserve, Limit globalReserve, long elapsedNanos) { if (isClosed) return; @@ -314,7 +314,7 @@ private void onReserveCapacityRegained(Limit endpointReserve, Limit globalReserv decoder.reactivate(); if (decoder.isActive()) - ClientMetrics.instance.unpauseConnection(); + onConnectionUnpaused(); } } catch (Throwable t) @@ -323,6 +323,10 @@ private void onReserveCapacityRegained(Limit endpointReserve, Limit globalReserv } } + protected void onConnectionUnpaused() + { + } + protected abstract void fatalExceptionCaught(Throwable t); // return true if the handler should be reactivated - if no new hurdles were encountered, diff --git a/src/java/org/apache/cassandra/transport/CQLMessageHandler.java b/src/java/org/apache/cassandra/transport/CQLMessageHandler.java index 2486497c2f37..2f764660a8c9 100644 --- a/src/java/org/apache/cassandra/transport/CQLMessageHandler.java +++ b/src/java/org/apache/cassandra/transport/CQLMessageHandler.java @@ -159,6 +159,12 @@ public boolean process(FrameDecoder.Frame frame) throws IOException return super.process(frame); } + @Override + protected void onConnectionUnpaused() + { + ClientMetrics.instance.unpauseConnection(); + } + /** * Checks limits on bytes in flight and the request rate limiter (if enabled), then takes one of three actions: * diff --git a/test/unit/org/apache/cassandra/net/AbstractMessageHandlerTest.java b/test/unit/org/apache/cassandra/net/AbstractMessageHandlerTest.java new file mode 100644 index 000000000000..901fc40f1595 --- /dev/null +++ b/test/unit/org/apache/cassandra/net/AbstractMessageHandlerTest.java @@ -0,0 +1,86 @@ +/* + * 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.io.IOException; +import java.util.concurrent.atomic.AtomicReference; + +import io.netty.channel.Channel; +import io.netty.channel.EventLoop; +import org.junit.Test; + +import org.apache.cassandra.net.ResourceLimits.Basic; +import org.apache.cassandra.net.ResourceLimits.Limit; + +import static org.junit.Assert.assertNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class AbstractMessageHandlerTest +{ + @Test + public void internodeReactivationDoesNotUseUninitializedClientMetrics() throws IOException + { + FrameDecoder decoder = mock(FrameDecoder.class); + when(decoder.isActive()).thenReturn(true); + + EventLoop eventLoop = mock(EventLoop.class); + when(eventLoop.inEventLoop()).thenReturn(true); + Channel channel = mock(Channel.class); + when(channel.eventLoop()).thenReturn(eventLoop); + + Limit endpointReserve = new Basic(1); + Limit globalReserve = new Basic(1); + AtomicReference failure = new AtomicReference<>(); + + InboundMessageHandler handler = new InboundMessageHandler(decoder, + ConnectionType.LARGE_MESSAGES, + channel, + null, + null, + MessagingService.current_version, + 1, + 1, + endpointReserve, + globalReserve, + AbstractMessageHandler.WaitQueue.endpoint(endpointReserve), + AbstractMessageHandler.WaitQueue.global(globalReserve), + ignored -> {}, + null, + null) + { + @Override + protected boolean processUpToOneMessage(Limit endpoint, Limit global) + { + return true; + } + + @Override + protected void fatalExceptionCaught(Throwable cause) + { + failure.set(cause); + } + }; + + handler.onReserveCapacityRegained(endpointReserve, globalReserve, 0); + + assertNull(failure.get()); + verify(decoder).reactivate(); + } +}