@Test public void testAppend() throws Exception { int partitionId = 0; int generation = 30; long sessionId = 20; long firstTransactionId = rand.nextInt(100); int seqNum = rand.nextInt(1000); ZooKeeperServerRunner zooKeeperServerRunner = new ZooKeeperServerRunner(0); try { String connectString = zooKeeperServerRunner.start(); ZooKeeperClient zkClient = new ZooKeeperClientImpl(connectString, 30000); ZNode root = zkClient.createPath(new ZNode("/test/store")); ZNode znode = new ZNode(root, Integer.toString(partitionId)); TestReplicaSessionManager replicaSessionManager = new TestReplicaSessionManager(1, NUM_REPLICAS); replicaSessionManager.setLastSessionInfo(partitionId, sessionId - 1L, -1L); replicaSessionManager.setMaxTransactionId(partitionId, firstTransactionId - 1); Map<ReplicaId, ReplicaState> replicaStates = new HashMap<>(); for (int i = 0; i < NUM_REPLICAS; i++) { ReplicaId replicaId = new ReplicaId(partitionId, String.format(TestReplicaSessionManager.CONNECT_STRING_TEMPLATE, i)); replicaStates.put(replicaId, new ReplicaState(replicaId, sessionId - 1L, ReplicaState.UNRESOLVED)); } zkClient.create( znode, new PartitionMetadata(generation - 1, sessionId, replicaStates), PartitionMetadataSerializer.INSTANCE, CreateMode.PERSISTENT ); try { ArrayList<ReplicaSession> replicaSessions = replicaSessionManager.getReplicaSessions(partitionId, sessionId); StoreSessionImpl session = new StoreSessionImpl( partitionId, generation, sessionId, WaltzServerConfig.DEFAULT_MAX_BATCH_SIZE, replicaSessions, zkClient, znode ); session.open(); try { try { Record record = TestUtils.record(new ReqId(1, generation - 1, partitionId, seqNum++), 0); TestUtils.syncAppend(session, record); fail(); } catch (GenerationMismatchException ex) { } Record[] records = new Record[10]; for (int i = 0; i < 10; i++) { records[i] = TestUtils.record(new ReqId(1, generation, partitionId, seqNum++), firstTransactionId + i); assertEquals(firstTransactionId + i, TestUtils.syncAppend(session, records[i])); assertEquals(firstTransactionId + i, session.highWaterMark()); } for (int i = 0; i < 10; i++) { long j = rand.nextInt(10); Record record = records[(int) j]; RecordHeader ret = session.getRecordHeader(firstTransactionId + j); if (ret == null) { System.out.println("null"); } assertEquals(new RecordHeader(firstTransactionId + j, record.reqId, record.header), ret); } for (int i = 0; i < 10; i++) { long j = rand.nextInt(10); assertEquals(records[(int) j], session.getRecord(firstTransactionId + j)); } assertNull(session.getRecordHeader(firstTransactionId + 10)); assertNull(session.getRecord(firstTransactionId + 10)); } finally { session.close(); } } finally { replicaSessionManager.close(); zkClient.close(); } } finally { zooKeeperServerRunner.stop(); zooKeeperServerRunner.clear(); } }
@Override public void append(StoreAppendRequest request) throws SessionClosedException, GenerationMismatchException { synchronized (this) { if (running) { if (request.reqId.generation() != generation) { throw new GenerationMismatchException(); } while (requestQueue.size() > maxBatchSize * 2) { try { wait(); } catch (InterruptedException ex) { Thread.interrupted(); } if (!running) { return; } } if (requestQueue.enqueue(request)) { numPending++; } } else { throw new SessionClosedException(); } } }
StoreSessionImpl implements StoreSession { @Override public void append(StoreAppendRequest request) throws SessionClosedException, GenerationMismatchException { synchronized (this) { if (running) { if (request.reqId.generation() != generation) { throw new GenerationMismatchException(); } while (requestQueue.size() > maxBatchSize * 2) { try { wait(); } catch (InterruptedException ex) { Thread.interrupted(); } if (!running) { return; } } if (requestQueue.enqueue(request)) { numPending++; } } else { throw new SessionClosedException(); } } } }
StoreSessionImpl implements StoreSession { @Override public void append(StoreAppendRequest request) throws SessionClosedException, GenerationMismatchException { synchronized (this) { if (running) { if (request.reqId.generation() != generation) { throw new GenerationMismatchException(); } while (requestQueue.size() > maxBatchSize * 2) { try { wait(); } catch (InterruptedException ex) { Thread.interrupted(); } if (!running) { return; } } if (requestQueue.enqueue(request)) { numPending++; } } else { throw new SessionClosedException(); } } } StoreSessionImpl(
        final int partitionId,
        final int generation,
        final long sessionId,
        final int maxBatchSize,
        final ArrayList<ReplicaSession> replicaSessions,
        final ZooKeeperClient zkClient,
        final ZNode znode
    ); }
StoreSessionImpl implements StoreSession { @Override public void append(StoreAppendRequest request) throws SessionClosedException, GenerationMismatchException { synchronized (this) { if (running) { if (request.reqId.generation() != generation) { throw new GenerationMismatchException(); } while (requestQueue.size() > maxBatchSize * 2) { try { wait(); } catch (InterruptedException ex) { Thread.interrupted(); } if (!running) { return; } } if (requestQueue.enqueue(request)) { numPending++; } } else { throw new SessionClosedException(); } } } StoreSessionImpl(
        final int partitionId,
        final int generation,
        final long sessionId,
        final int maxBatchSize,
        final ArrayList<ReplicaSession> replicaSessions,
        final ZooKeeperClient zkClient,
        final ZNode znode
    ); void open(); @Override void close(); @Override int generation(); @Override long highWaterMark(); @Override long lowWaterMark(); @Override boolean isWritable(); @Override void append(StoreAppendRequest request); @Override int numPendingAppends(); @Override void resolveAllAppendRequests(long highWaterMark); @Override long flush(); @Override Record getRecord(long transactionId); @Override ArrayList<Record> getRecordList(long transactionId, int maxNumRecords); @Override RecordHeader getRecordHeader(long transactionId); @Override ArrayList<RecordHeader> getRecordHeaderList(long transactionId, int maxNumRecords); }
StoreSessionImpl implements StoreSession { @Override public void append(StoreAppendRequest request) throws SessionClosedException, GenerationMismatchException { synchronized (this) { if (running) { if (request.reqId.generation() != generation) { throw new GenerationMismatchException(); } while (requestQueue.size() > maxBatchSize * 2) { try { wait(); } catch (InterruptedException ex) { Thread.interrupted(); } if (!running) { return; } } if (requestQueue.enqueue(request)) { numPending++; } } else { throw new SessionClosedException(); } } } StoreSessionImpl(
        final int partitionId,
        final int generation,
        final long sessionId,
        final int maxBatchSize,
        final ArrayList<ReplicaSession> replicaSessions,
        final ZooKeeperClient zkClient,
        final ZNode znode
    ); void open(); @Override void close(); @Override int generation(); @Override long highWaterMark(); @Override long lowWaterMark(); @Override boolean isWritable(); @Override void append(StoreAppendRequest request); @Override int numPendingAppends(); @Override void resolveAllAppendRequests(long highWaterMark); @Override long flush(); @Override Record getRecord(long transactionId); @Override ArrayList<Record> getRecordList(long transactionId, int maxNumRecords); @Override RecordHeader getRecordHeader(long transactionId); @Override ArrayList<RecordHeader> getRecordHeaderList(long transactionId, int maxNumRecords); final int generation; final int partitionId; final long sessionId; }