### Question:
PrepRequestProcessor extends ZooKeeperCriticalThread implements RequestProcessor { protected void pRequest(Request request) throws RequestProcessorException { request.setHdr(null); request.setTxn(null); if (!request.isThrottled()) { pRequestHelper(request); } request.zxid = zks.getZxid(); long timeFinishedPrepare = Time.currentElapsedTime(); ServerMetrics.getMetrics().PREP_PROCESS_TIME.add(timeFinishedPrepare - request.prepStartTime); nextProcessor.processRequest(request); ServerMetrics.getMetrics().PROPOSAL_PROCESS_TIME.add(Time.currentElapsedTime() - timeFinishedPrepare); } PrepRequestProcessor(ZooKeeperServer zks, RequestProcessor nextProcessor); static void setFailCreate(boolean b); @Override void run(); static List<ACL> fixupACL(String path, List<Id> authInfo, List<ACL> acls); void processRequest(Request request); void shutdown();  }### Answer:
@Test public void testPRequest() throws Exception { pLatch = new CountDownLatch(1); processor = new PrepRequestProcessor(zks, new MyRequestProcessor()); Request foo = new Request(null, 1L, 1, OpCode.create, ByteBuffer.allocate(3), null); processor.pRequest(foo); assertEquals(new ErrorTxn(KeeperException.Code.MARSHALLINGERROR.intValue()), outcome.getTxn(), "Request should have marshalling error"); assertTrue(pLatch.await(5, TimeUnit.SECONDS), "request hasn't been processed in chain"); }
@Test public void testReconfigWithAnotherOutstandingChange() throws Exception { QuorumPeerConfig.setReconfigEnabled(true); QuorumPeerConfig.setStandaloneEnabled(false); QuorumPeer qp = new QuorumPeer(); QuorumVerifier quorumVerifierMock = mock(QuorumVerifier.class); when(quorumVerifierMock.getAllMembers()).thenReturn(LeaderBeanTest.getMockedPeerViews(qp.getId())); qp.setQuorumVerifier(quorumVerifierMock, false); FileTxnSnapLog snapLog = new FileTxnSnapLog(tmpDir, tmpDir); LeaderZooKeeperServer lzks = new LeaderZooKeeperServer(snapLog, qp, new ZKDatabase(snapLog)); qp.leader = new Leader(qp, lzks); lzks.sessionTracker = new MySessionTracker(); ZooKeeperServer.setDigestEnabled(true); processor = new PrepRequestProcessor(lzks, new MyRequestProcessor()); Record record = new CreateRequest("/foo", "data".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT.toFlag()); pLatch = new CountDownLatch(1); processor.pRequest(createRequest(record, OpCode.create, false)); assertTrue(pLatch.await(5, TimeUnit.SECONDS), "request hasn't been processed in chain"); String newMember = "server.0=localhost:" + PortAssignment.unique() + ":" + PortAssignment.unique() + ":participant"; record = new ReconfigRequest(null, null, newMember, 0); pLatch = new CountDownLatch(1); processor.pRequest(createRequest(record, OpCode.reconfig, true)); assertTrue(pLatch.await(5, TimeUnit.SECONDS), "request hasn't been processed in chain"); assertEquals(outcome.getHdr().getType(), OpCode.reconfig); }
@Test public void testInvalidPath() throws Exception { pLatch = new CountDownLatch(1); processor = new PrepRequestProcessor(zks, new MyRequestProcessor()); SetDataRequest record = new SetDataRequest("", new byte[0], -1); Request req = createRequest(record, OpCode.setData, false); processor.pRequest(req); pLatch.await(); assertEquals(outcome.getHdr().getType(), OpCode.error); assertEquals(outcome.getException().code(), KeeperException.Code.BADARGUMENTS); }