ZOOKEEPER-4717: Cache serialize data in the request to avoid repeat serialize. (#2030)

zookeeper-server/src/main/java/org/apache/zookeeper/server/Request.java

@@ -19,6 +19,7 @@
 package org.apache.zookeeper.server;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.List;
@@ -30,11 +31,14 @@ import org.apache.zookeeper.common.Time;
 import org.apache.zookeeper.data.Id;
 import org.apache.zookeeper.metrics.Summary;
 import org.apache.zookeeper.metrics.SummarySet;
+import org.apache.zookeeper.server.persistence.Util;
 import org.apache.zookeeper.server.quorum.LearnerHandler;
 import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
 import org.apache.zookeeper.server.util.AuthUtil;
 import org.apache.zookeeper.txn.TxnDigest;
 import org.apache.zookeeper.txn.TxnHeader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * This is the structure that represents a request moving through a chain of
@@ -42,6 +46,7 @@ import org.apache.zookeeper.txn.TxnHeader;
  * onto the request as it is processed.
  */
 public class Request {
+    private static final Logger LOG = LoggerFactory.getLogger(Request.class);
 
     public static final Request requestOfDeath = new Request(null, 0, 0, 0, null, null);
 
@@ -164,6 +169,26 @@ public class Request {
                 && this.type != OpCode.createSession;
     }
 
+    private transient byte[] serializeData;
+
+    @SuppressFBWarnings(value = "EI_EXPOSE_REP")
+    public byte[] getSerializeData() {
+        if (this.hdr == null) {
+            return null;
+        }
+
+        if (this.serializeData == null) {
+            try {
+                this.serializeData = Util.marshallTxnEntry(this.hdr, this.txn, this.txnDigest);
+            } catch (IOException e) {
+                LOG.error("This really should be impossible.", e);
+                this.serializeData = new byte[32];
+            }
+        }
+
+        return this.serializeData;
+    }
+
     /**
      * If this is a create or close request for a local-only session.
      */

zookeeper-server/src/main/java/org/apache/zookeeper/server/ZKDatabase.java

@@ -329,8 +329,7 @@ public class ZKDatabase {
                 minCommittedLog = request.zxid;
                 maxCommittedLog = request.zxid;
             }
-
-            byte[] data = SerializeUtils.serializeRequest(request);
+            byte[] data = request.getSerializeData();
             QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid, data, null);
             Proposal p = new Proposal();
             p.packet = pp;

zookeeper-server/src/main/java/org/apache/zookeeper/server/persistence/FileTxnLog.java

@@ -257,18 +257,9 @@ public class FileTxnLog implements TxnLog, Closeable {
         }
     }
 
-    /**
-     * append an entry to the transaction log
-     * @param hdr the header of the transaction
-     * @param txn the transaction part of the entry
-     * returns true iff something appended, otw false
-     */
-    public synchronized boolean append(TxnHeader hdr, Record txn) throws IOException {
-              return append(hdr, txn, null);
-    }
-
     @Override
-    public synchronized boolean append(TxnHeader hdr, Record txn, TxnDigest digest) throws IOException {
+    public synchronized boolean append(Request request) throws IOException {
+        TxnHeader hdr = request.getHdr();
         if (hdr == null) {
             return false;
         }
@@ -296,7 +287,7 @@ public class FileTxnLog implements TxnLog, Closeable {
             streamsToFlush.add(fos);
         }
         filePadding.padFile(fos.getChannel());
-        byte[] buf = Util.marshallTxnEntry(hdr, txn, digest);
+        byte[] buf = request.getSerializeData();
         if (buf == null || buf.length == 0) {
             throw new IOException("Faulty serialization for header " + "and txn");
         }

zookeeper-server/src/main/java/org/apache/zookeeper/server/persistence/FileTxnSnapLog.java

@@ -589,7 +589,7 @@ public class FileTxnSnapLog {
      * @throws IOException
      */
     public boolean append(Request si) throws IOException {
-        return txnLog.append(si.getHdr(), si.getTxn(), si.getTxnDigest());
+        return txnLog.append(si);
     }
 
     /**

zookeeper-server/src/main/java/org/apache/zookeeper/server/persistence/TxnLog.java

@@ -21,6 +21,7 @@ package org.apache.zookeeper.server.persistence;
 import java.io.Closeable;
 import java.io.IOException;
 import org.apache.jute.Record;
+import org.apache.zookeeper.server.Request;
 import org.apache.zookeeper.server.ServerStats;
 import org.apache.zookeeper.txn.TxnDigest;
 import org.apache.zookeeper.txn.TxnHeader;
@@ -43,24 +44,14 @@ public interface TxnLog extends Closeable {
      * @throws IOException
      */
     void rollLog() throws IOException;
-    /**
-     * Append a request to the transaction log
-     * @param hdr the transaction header
-     * @param r the transaction itself
-     * @return true iff something appended, otw false
-     * @throws IOException
-     */
-    boolean append(TxnHeader hdr, Record r) throws IOException;
 
     /**
      * Append a request to the transaction log with a digset
-     * @param hdr the transaction header
-     * @param r the transaction itself
-     * @param digest transaction digest
+     * @param request the request to be appended
      * returns true iff something appended, otw false
      * @throws IOException
      */
-    boolean append(TxnHeader hdr, Record r, TxnDigest digest) throws IOException;
+    boolean append(Request request) throws IOException;
 
     /**
      * Start reading the transaction logs

zookeeper-server/src/main/java/org/apache/zookeeper/server/persistence/Util.java

@@ -172,23 +172,19 @@ public class Util {
         return null;
     }
 
+
     /**
      * Serializes transaction header and transaction data into a byte buffer.
      *
      * @param hdr transaction header
      * @param txn transaction data
+     * @param digest transaction digest
+     *
      * @return serialized transaction record
-     * @throws IOException
      */
-    public static byte[] marshallTxnEntry(TxnHeader hdr, Record txn) throws IOException {
-        return marshallTxnEntry(hdr, txn, null);
-    }
-
-    public static byte[] marshallTxnEntry(TxnHeader hdr, Record txn, TxnDigest digest)
-            throws IOException {
+    public static byte[] marshallTxnEntry(TxnHeader hdr, Record txn, TxnDigest digest) throws IOException {
         ByteArrayOutputStream baos = new ByteArrayOutputStream();
         OutputArchive boa = BinaryOutputArchive.getArchive(baos);
-
         hdr.serialize(boa, "hdr");
         if (txn != null) {
             txn.serialize(boa, "txn");

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Leader.java

@@ -68,7 +68,6 @@ import org.apache.zookeeper.server.ZooTrace;
 import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
 import org.apache.zookeeper.server.quorum.auth.QuorumAuthServer;
 import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
-import org.apache.zookeeper.server.util.SerializeUtils;
 import org.apache.zookeeper.server.util.ZxidUtils;
 import org.apache.zookeeper.util.ServiceUtils;
 import org.slf4j.Logger;
@@ -1251,7 +1250,7 @@ public class Leader extends LearnerMaster {
             throw new XidRolloverException(msg);
         }
 
-        byte[] data = SerializeUtils.serializeRequest(request);
+        byte[] data = request.getSerializeData();
         proposalStats.setLastBufferSize(data.length);
         QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid, data, null);
 

zookeeper-server/src/main/java/org/apache/zookeeper/server/util/SerializeUtils.java

@@ -30,11 +30,9 @@ import org.apache.jute.OutputArchive;
 import org.apache.jute.Record;
 import org.apache.zookeeper.ZooDefs.OpCode;
 import org.apache.zookeeper.server.DataTree;
-import org.apache.zookeeper.server.Request;
 import org.apache.zookeeper.server.TxnLogEntry;
 import org.apache.zookeeper.server.ZooKeeperServer;
 import org.apache.zookeeper.server.ZooTrace;
-import org.apache.zookeeper.server.persistence.Util;
 import org.apache.zookeeper.txn.CloseSessionTxn;
 import org.apache.zookeeper.txn.CreateContainerTxn;
 import org.apache.zookeeper.txn.CreateSessionTxn;
@@ -170,19 +168,4 @@ public class SerializeUtils {
         }
         dt.serialize(oa, "tree");
     }
-
-    public static byte[] serializeRequest(Request request) {
-        if (request == null || request.getHdr() == null) {
-            return null;
-        }
-        byte[] data;
-        try {
-            data = Util.marshallTxnEntry(request.getHdr(), request.getTxn(), request.getTxnDigest());
-        } catch (IOException e) {
-            LOG.error("This really should be impossible", e);
-            data = new byte[32];
-        }
-        return data;
-    }
-
 }

zookeeper-server/src/test/java/org/apache/zookeeper/server/TxnLogDigestTest.java

@@ -31,7 +31,6 @@ import java.util.Map;
 import mockit.Invocation;
 import mockit.Mock;
 import mockit.MockUp;
-import org.apache.jute.Record;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.Op;
 import org.apache.zookeeper.ZooDefs;
@@ -265,13 +264,13 @@ public class TxnLogDigestTest extends ClientBase {
         static long skipAppendZxid = -1;
 
         @Mock
-        public synchronized boolean append(Invocation invocation, TxnHeader hdr,
-                Record txn, TxnDigest digest) throws IOException {
+        public synchronized boolean append(Invocation invocation, Request request) throws IOException {
+            TxnHeader hdr = request.getHdr();
             if (hdr != null && hdr.getZxid() == skipAppendZxid) {
                 LOG.info("skipping txn {}", skipAppendZxid);
                 return true;
             }
-            return invocation.proceed(hdr, txn, digest);
+            return invocation.proceed(request);
         }
 
         public static void reset() {

zookeeper-server/src/test/java/org/apache/zookeeper/server/persistence/FileTxnLogTest.java

@@ -39,6 +39,7 @@ import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.data.Stat;
 import org.apache.zookeeper.proto.CreateRequest;
+import org.apache.zookeeper.server.Request;
 import org.apache.zookeeper.server.ServerCnxnFactory;
 import org.apache.zookeeper.server.ServerStats;
 import org.apache.zookeeper.server.ZKDatabase;
@@ -98,13 +99,19 @@ public class FileTxnLogTest extends ZKTestCase {
 
         // Append and commit 2 transactions to the log
         // Prior to ZOOKEEPER-2249, attempting to pad in association with the second transaction will corrupt the first
-        fileTxnLog.append(
-            new TxnHeader(1, 1, 1, 1, ZooDefs.OpCode.create),
-            new CreateTxn("/testPreAllocSizeSmallerThanTxnData1", data, ZooDefs.Ids.OPEN_ACL_UNSAFE, false, 0));
+
+
+        fileTxnLog.append(new Request(0, 0, 0,
+                new TxnHeader(1, 1, 1, 1, ZooDefs.OpCode.create),
+                new CreateTxn("/testPreAllocSizeSmallerThanTxnData1", data, ZooDefs.Ids.OPEN_ACL_UNSAFE, false, 0),
+                0));
         fileTxnLog.commit();
-        fileTxnLog.append(
-            new TxnHeader(1, 1, 2, 2, ZooDefs.OpCode.create),
-            new CreateTxn("/testPreAllocSizeSmallerThanTxnData2", new byte[]{}, ZooDefs.Ids.OPEN_ACL_UNSAFE, false, 0));
+
+        fileTxnLog.append(new Request(0, 0, 0,
+                new TxnHeader(1, 1, 2, 2, ZooDefs.OpCode.create),
+                new CreateTxn("/testPreAllocSizeSmallerThanTxnData2", new byte[]{},
+                        ZooDefs.Ids.OPEN_ACL_UNSAFE, false, 0),
+                0));
         fileTxnLog.commit();
         fileTxnLog.close();
 
@@ -143,11 +150,14 @@ public class FileTxnLogTest extends ZKTestCase {
         // Verify serverStats is 0 before any commit
         assertEquals(0L, serverStats.getFsyncThresholdExceedCount());
 
+
         // When ...
         for (int i = 0; i < 50; i++) {
-            fileTxnLog.append(
-                new TxnHeader(1, 1, 1, 1, ZooDefs.OpCode.create),
-                new CreateTxn("/testFsyncThresholdCountIncreased", new byte[]{}, ZooDefs.Ids.OPEN_ACL_UNSAFE, false, 0));
+            fileTxnLog.append(new Request(0, 0, 0,
+                    new TxnHeader(1, 1, 1, 1, ZooDefs.OpCode.create),
+                    new CreateTxn("/testFsyncThresholdCountIncreased", new byte[]{},
+                            ZooDefs.Ids.OPEN_ACL_UNSAFE, false, 0),
+                    0));
             fileTxnLog.commit(); // only 1 commit, otherwise it will be flaky
             // Then ... verify serverStats is updated to the number of commits (as threshold is set to 0)
             assertEquals((long) i + 1, serverStats.getFsyncThresholdExceedCount());
@@ -181,7 +191,8 @@ public class FileTxnLogTest extends ZKTestCase {
                 logSize += fileHeaderSize;
                 position += fileHeaderSize;
             }
-            log.append(new TxnHeader(0, 0, zxid++, 0, 0), record);
+
+            log.append(new Request(0, 0, 0, new TxnHeader(0, 0, zxid++, 0, 0), record, 0));
             logSize += PREALLOCATE;
             assertEquals(logSize, log.getCurrentLogSize());
             assertEquals(position, log.fos.getChannel().position());
@@ -196,7 +207,7 @@ public class FileTxnLogTest extends ZKTestCase {
         position = totalSize;
         boolean recalculate = true;
         for (int i = 0; i < 4; i++) {
-            log.append(new TxnHeader(0, 0, zxid++, 0, 0), record);
+            log.append(new Request(0, 0, 0, new TxnHeader(0, 0, zxid++, 0, 0), record, 0));
             if (recalculate) {
                 recalculate = false;
             } else {
@@ -290,7 +301,7 @@ public class FileTxnLogTest extends ZKTestCase {
     private int calculateSingleRecordLength(TxnHeader txnHeader, Record record) throws IOException {
         int crcLength = 8;
         int dataLength = 4;
-        int recordLength = Util.marshallTxnEntry(txnHeader, record).length;
+        int recordLength = Util.marshallTxnEntry(txnHeader, record, null).length;
         int endFlagLength = 1;
         return crcLength + dataLength + recordLength + endFlagLength;
     }

zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LeaderBeanTest.java

@@ -42,7 +42,6 @@ import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
 import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
 import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
 import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
-import org.apache.zookeeper.server.util.SerializeUtils;
 import org.apache.zookeeper.test.ClientBase;
 import org.apache.zookeeper.txn.TxnHeader;
 import org.junit.jupiter.api.AfterEach;
@@ -125,7 +124,7 @@ public class LeaderBeanTest {
         leader.propose(req);
 
         // Assert
-        byte[] data = SerializeUtils.serializeRequest(req);
+        byte[] data = req.getSerializeData();
         assertEquals(data.length, leaderBean.getLastProposalSize());
         assertEquals(data.length, leaderBean.getMinProposalSize());
         assertEquals(data.length, leaderBean.getMaxProposalSize());

zookeeper-server/src/test/java/org/apache/zookeeper/server/util/SerializeUtilsTest.java

@@ -42,16 +42,10 @@ import org.mockito.stubbing.Answer;
 
 public class SerializeUtilsTest {
 
-    @Test
-    public void testSerializeRequestRequestIsNull() {
-        byte[] data = SerializeUtils.serializeRequest(null);
-        assertNull(data);
-    }
-
     @Test
     public void testSerializeRequestRequestHeaderIsNull() {
         Request request = new Request(0, 0, 0, null, null, 0);
-        byte[] data = SerializeUtils.serializeRequest(request);
+        byte[] data = request.getSerializeData();
         assertNull(data);
     }
 
@@ -71,7 +65,7 @@ public class SerializeUtilsTest {
         Request request = new Request(1, 2, 3, header, null, 4);
 
         // Act
-        byte[] data = SerializeUtils.serializeRequest(request);
+        byte[] data = request.getSerializeData();
 
         // Assert
         assertNotNull(data);
@@ -109,7 +103,7 @@ public class SerializeUtilsTest {
         Request request = new Request(1, 2, 3, header, txn, 4);
 
         // Act
-        byte[] data = SerializeUtils.serializeRequest(request);
+        byte[] data = request.getSerializeData();
 
         // Assert
         assertNotNull(data);

zookeeper-server/src/test/java/org/apache/zookeeper/test/LoadFromLogNoServerTest.java

@@ -34,6 +34,7 @@ import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.common.Time;
 import org.apache.zookeeper.server.DataNode;
 import org.apache.zookeeper.server.DataTree;
+import org.apache.zookeeper.server.Request;
 import org.apache.zookeeper.server.ZooKeeperServer;
 import org.apache.zookeeper.server.persistence.FileHeader;
 import org.apache.zookeeper.server.persistence.FileTxnLog;
@@ -166,7 +167,7 @@ public class LoadFromLogNoServerTest extends ZKTestCase {
         FileTxnLog txnLog = new FileTxnLog(tmpDir);
         TxnHeader txnHeader = new TxnHeader(0xabcd, 0x123, 0x123, Time.currentElapsedTime(), ZooDefs.OpCode.create);
         Record txn = new CreateTxn("/Test", new byte[0], null, false, 1);
-        txnLog.append(txnHeader, txn);
+        txnLog.append(new Request(0, 0, 0, txnHeader, txn, 0));
         FileInputStream in = new FileInputStream(tmpDir.getPath() + "/log." + Long.toHexString(txnHeader.getZxid()));
         BinaryInputArchive ia = BinaryInputArchive.getArchive(in);
         FileHeader header = new FileHeader();

zookeeper-server/src/test/java/org/apache/zookeeper/test/LogChopperTest.java

@@ -31,6 +31,7 @@ import java.nio.file.Path;
 import java.nio.file.SimpleFileVisitor;
 import java.nio.file.attribute.BasicFileAttributes;
 import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.server.Request;
 import org.apache.zookeeper.server.persistence.FileTxnLog;
 import org.apache.zookeeper.server.persistence.TxnLog;
 import org.apache.zookeeper.server.util.LogChopper;
@@ -101,12 +102,12 @@ public class LogChopperTest extends ClientBase {
 
         for (int i = 0; i < 100; i++) {
             TxnHeader hdr = new TxnHeader(clientId, cxid, ++zxid, ++time, type);
-            txnLog.append(hdr, txn);
+            txnLog.append(new Request(0, 0, 0, hdr, txn, 0));
         }
 
         // append a txn with gap
         TxnHeader hdr = new TxnHeader(clientId, cxid, zxid + 10, ++time, type);
-        txnLog.append(hdr, txn);
+        txnLog.append(new Request(0, 0, 0, hdr, txn, 0));
 
         txnLog.commit();