You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@bookkeeper.apache.org by si...@apache.org on 2018/03/10 08:28:12 UTC
[bookkeeper] branch branch-4.6 updated: Disable using v2 decoder to
decode v3 response if usingV2WireProtocol is enabled and handle unknown op
in v2 request decoder
This is an automated email from the ASF dual-hosted git repository.
sijie pushed a commit to branch branch-4.6
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
The following commit(s) were added to refs/heads/branch-4.6 by this push:
new 3a99922 Disable using v2 decoder to decode v3 response if usingV2WireProtocol is enabled and handle unknown op in v2 request decoder
3a99922 is described below
commit 3a999226e10936951c7081df3b6d8b54204dfdc2
Author: Sijie Guo <si...@apache.org>
AuthorDate: Sat Mar 10 00:13:57 2018 -0800
Disable using v2 decoder to decode v3 response if usingV2WireProtocol is enabled and handle unknown op in v2 request decoder
Descriptions of the changes in this PR:
*Problem*
- response decoder should use whatever protocol that request is using to decode the response. otherwise it is making wrong assumption on response protocol.
- v2 request decoder doesn't handle unknown op correctly, which will cause netty double-release bytebuf
*Solution*
- make sure response decoder respect the protocol that client is using for sending requests
- v2 request decoder should throw exceptions when fail to decode unknown request
Related issues: #198
Author: Sijie Guo <si...@apache.org>
Reviewers: Ivan Kelly <iv...@apache.org>, Enrico Olivelli <eo...@gmail.com>, Jia Zhai <None>, Matteo Merli <mm...@apache.org>
This closes #1240 from sijie/bookkeeper_fix_bytebuf_issues
(cherry picked from commit 59e48a390c1caa12adb7cc89afd2f9d48790a674)
Signed-off-by: Sijie Guo <si...@apache.org>
---
.../bookkeeper/proto/BookieProtoEncoding.java | 21 ++--
.../bookkeeper/proto/PerChannelBookieClient.java | 4 +-
.../bookkeeper/proto/BookieProtoEncodingTest.java | 121 +++++++++++++++++++++
.../bookkeeper/proto/TestBackwardCompatCMS42.java | 51 ++++++---
4 files changed, 165 insertions(+), 32 deletions(-)
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieProtoEncoding.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieProtoEncoding.java
index 4f2b142..af17ffa 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieProtoEncoding.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieProtoEncoding.java
@@ -178,9 +178,10 @@ public class BookieProtoEncoding {
= BookkeeperProtocol.AuthMessage.newBuilder();
builder.mergeFrom(new ByteBufInputStream(packet), extensionRegistry);
return new BookieProtocol.AuthRequest(version, builder.build());
- }
- return packet;
+ default:
+ throw new IllegalStateException("Received unknown request op code = " + opCode);
+ }
}
private static byte[] readMasterKey(ByteBuf packet) {
@@ -446,12 +447,12 @@ public class BookieProtoEncoding {
public static class ResponseDecoder extends MessageToMessageDecoder<Object> {
final EnDecoder REP_PREV3;
final EnDecoder REP_V3;
- boolean usingV3Protocol;
+ boolean usingV2Protocol;
- ResponseDecoder(ExtensionRegistry extensionRegistry) {
+ ResponseDecoder(ExtensionRegistry extensionRegistry, boolean useV2Protocol) {
REP_PREV3 = new ResponseEnDeCoderPreV3(extensionRegistry);
REP_V3 = new ResponseEnDecoderV3(extensionRegistry);
- usingV3Protocol = true;
+ usingV2Protocol = useV2Protocol;
}
@Override
@@ -465,14 +466,8 @@ public class BookieProtoEncoding {
ByteBuf buffer = (ByteBuf) msg;
buffer.markReaderIndex();
- if (usingV3Protocol) {
- try {
- out.add(REP_V3.decode(buffer));
- } catch (InvalidProtocolBufferException e) {
- usingV3Protocol = false;
- buffer.resetReaderIndex();
- out.add(REP_PREV3.decode(buffer));
- }
+ if (!usingV2Protocol) {
+ out.add(REP_PREV3.decode(buffer));
} else {
// If in the same connection we already got preV3 messages, don't try again to decode V3 messages
out.add(REP_PREV3.decode(buffer));
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
index 3d9ba13..2317a7c 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
@@ -415,7 +415,9 @@ public class PerChannelBookieClient extends ChannelInboundHandlerAdapter {
new LengthFieldBasedFrameDecoder(maxFrameSize, 0, 4, 0, 4));
pipeline.addLast("lengthprepender", new LengthFieldPrepender(4));
pipeline.addLast("bookieProtoEncoder", new BookieProtoEncoding.RequestEncoder(extRegistry));
- pipeline.addLast("bookieProtoDecoder", new BookieProtoEncoding.ResponseDecoder(extRegistry));
+ pipeline.addLast(
+ "bookieProtoDecoder",
+ new BookieProtoEncoding.ResponseDecoder(extRegistry, useV2WireProtocol));
pipeline.addLast("authHandler", new AuthHandler.ClientSideHandler(authProviderFactory, txnIdGenerator, connectionPeer));
pipeline.addLast("mainhandler", PerChannelBookieClient.this);
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/BookieProtoEncodingTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/BookieProtoEncodingTest.java
new file mode 100644
index 0000000..f182371
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/BookieProtoEncodingTest.java
@@ -0,0 +1,121 @@
+/*
+ * 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.bookkeeper.proto;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+
+import com.google.common.collect.Lists;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.UnpooledByteBufAllocator;
+import io.netty.channel.ChannelHandlerContext;
+import java.util.List;
+import org.apache.bookkeeper.proto.BookieProtoEncoding.RequestEnDeCoderPreV3;
+import org.apache.bookkeeper.proto.BookieProtoEncoding.RequestEnDecoderV3;
+import org.apache.bookkeeper.proto.BookieProtoEncoding.ResponseDecoder;
+import org.apache.bookkeeper.proto.BookieProtoEncoding.ResponseEnDeCoderPreV3;
+import org.apache.bookkeeper.proto.BookieProtoEncoding.ResponseEnDecoderV3;
+import org.apache.bookkeeper.proto.BookieProtocol.AddResponse;
+import org.apache.bookkeeper.proto.BookkeeperProtocol.AddRequest.Flag;
+import org.apache.bookkeeper.proto.BookkeeperProtocol.BKPacketHeader;
+import org.apache.bookkeeper.proto.BookkeeperProtocol.OperationType;
+import org.apache.bookkeeper.proto.BookkeeperProtocol.ProtocolVersion;
+import org.apache.bookkeeper.proto.BookkeeperProtocol.StatusCode;
+import org.junit.Test;
+
+/**
+ * Unit test {@link BookieProtoEncoding}.
+ */
+public class BookieProtoEncodingTest {
+
+ @Test
+ public void testV3ResponseDecoderNoFallback() throws Exception {
+ AddResponse v2Resp = AddResponse.create(
+ BookieProtocol.CURRENT_PROTOCOL_VERSION,
+ BookieProtocol.EOK,
+ 1L,
+ 2L);
+
+ BookkeeperProtocol.Response v3Resp = BookkeeperProtocol.Response.newBuilder()
+ .setHeader(BKPacketHeader.newBuilder()
+ .setVersion(ProtocolVersion.VERSION_THREE)
+ .setTxnId(1L)
+ .setOperation(OperationType.ADD_ENTRY)
+ .build())
+ .setStatus(StatusCode.EOK)
+ .setAddResponse(BookkeeperProtocol.AddResponse.newBuilder()
+ .setStatus(StatusCode.EOK)
+ .setLedgerId(1L)
+ .setEntryId(2L)
+ .build())
+ .build();
+
+ List<Object> outList = Lists.newArrayList();
+
+ ResponseEnDeCoderPreV3 v2Encoder = new ResponseEnDeCoderPreV3(null);
+ ResponseEnDecoderV3 v3Encoder = new ResponseEnDecoderV3(null);
+
+ ResponseDecoder v3Decoder = new ResponseDecoder(null, false);
+ try {
+ v3Decoder.decode(
+ mock(ChannelHandlerContext.class),
+ v2Encoder.encode(v2Resp, UnpooledByteBufAllocator.DEFAULT),
+ outList
+ );
+ fail("V3 response decoder should fail on decoding v2 response");
+ } catch (InvalidProtocolBufferException e) {
+ // expected
+ }
+ assertEquals(0, outList.size());
+
+ v3Decoder.decode(
+ mock(ChannelHandlerContext.class),
+ v3Encoder.encode(v3Resp, UnpooledByteBufAllocator.DEFAULT),
+ outList);
+ assertEquals(1, outList.size());
+ }
+
+ @Test(expected = IllegalStateException.class)
+ public void testV2RequestDecoderThrowExceptionOnUnknownRequests() throws Exception {
+ RequestEnDeCoderPreV3 v2ReqEncoder = new RequestEnDeCoderPreV3(null);
+ RequestEnDecoderV3 v3ReqEncoder = new RequestEnDecoderV3(null);
+
+ BookkeeperProtocol.Request v3Req = BookkeeperProtocol.Request.newBuilder()
+ .setHeader(BKPacketHeader.newBuilder()
+ .setVersion(ProtocolVersion.VERSION_THREE)
+ .setTxnId(1L)
+ .setOperation(OperationType.ADD_ENTRY)
+ .build())
+ .setAddRequest(BookkeeperProtocol.AddRequest.newBuilder()
+ .setLedgerId(1L)
+ .setEntryId(2L)
+ .setMasterKey(ByteString.copyFrom("", UTF_8))
+ .setFlag(Flag.RECOVERY_ADD)
+ .setBody(ByteString.copyFrom("test", UTF_8)))
+ .build();
+
+
+ v2ReqEncoder.decode((ByteBuf) v3ReqEncoder.encode(v3Req, UnpooledByteBufAllocator.DEFAULT));
+ }
+
+}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/TestBackwardCompatCMS42.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/TestBackwardCompatCMS42.java
index 4619a92..89c4517 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/TestBackwardCompatCMS42.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/TestBackwardCompatCMS42.java
@@ -20,13 +20,11 @@
*/
package org.apache.bookkeeper.proto;
-import com.google.protobuf.ByteString;
-import org.apache.bookkeeper.conf.ServerConfiguration;
-import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import com.google.protobuf.ByteString;
import com.google.protobuf.ExtensionRegistry;
import io.netty.channel.Channel;
@@ -34,23 +32,27 @@ import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ArrayBlockingQueue;
+
import org.apache.bookkeeper.auth.ClientAuthProvider;
import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.net.BookieSocketAddress;
-import org.apache.bookkeeper.util.OrderedSafeExecutor;
import org.apache.bookkeeper.auth.TestAuth;
import org.apache.bookkeeper.auth.AuthProviderFactoryFactory;
-
-import org.apache.bookkeeper.proto.BookieProtocol.*;
+import org.apache.bookkeeper.proto.BookieProtocol.AuthRequest;
+import org.apache.bookkeeper.proto.BookieProtocol.AuthResponse;
+import org.apache.bookkeeper.proto.BookieProtocol.ReadRequest;
+import org.apache.bookkeeper.proto.BookieProtocol.Request;
+import org.apache.bookkeeper.proto.BookieProtocol.Response;
import org.apache.bookkeeper.proto.BookkeeperProtocol.AuthMessage;
-
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ArrayBlockingQueue;
-
-import static org.junit.Assert.*;
+import org.apache.bookkeeper.stats.NullStatsLogger;
+import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
+import org.apache.bookkeeper.util.OrderedSafeExecutor;
+import org.junit.Test;
public class TestBackwardCompatCMS42 extends BookKeeperClusterTestCase {
- static final Logger LOG = LoggerFactory.getLogger(TestBackwardCompatCMS42.class);
private static final byte[] SUCCESS_RESPONSE = {1};
private static final byte[] FAILURE_RESPONSE = {2};
@@ -174,7 +176,9 @@ public class TestBackwardCompatCMS42 extends BookKeeperClusterTestCase {
}
CompatClient42 newCompatClient(BookieSocketAddress addr) throws Exception {
- return new CompatClient42(executor, eventLoopGroup, addr, authProvider, extRegistry);
+ ClientConfiguration conf = new ClientConfiguration();
+ conf.setUseV2WireProtocol(true);
+ return new CompatClient42(conf, executor, eventLoopGroup, addr, authProvider, extRegistry);
}
// extending PerChannelBookieClient to get the pipeline factory
@@ -183,11 +187,22 @@ public class TestBackwardCompatCMS42 extends BookKeeperClusterTestCase {
Channel channel;
final CountDownLatch connected = new CountDownLatch(1);
- CompatClient42(OrderedSafeExecutor executor, EventLoopGroup eventLoopGroup,
+ CompatClient42(ClientConfiguration conf,
+ OrderedSafeExecutor executor,
+ EventLoopGroup eventLoopGroup,
BookieSocketAddress addr,
ClientAuthProvider.Factory authProviderFactory,
ExtensionRegistry extRegistry) throws Exception {
- super(executor, eventLoopGroup, addr, authProviderFactory, extRegistry);
+ super(
+ conf,
+ executor,
+ eventLoopGroup,
+ addr,
+ null,
+ NullStatsLogger.INSTANCE,
+ authProviderFactory,
+ extRegistry,
+ null);
state = ConnectionState.CONNECTING;
ChannelFuture future = connect();
--
To stop receiving notification emails like this one, please contact
sijie@apache.org.