@Test public void testCatchUp() throws Exception { Properties properties = new Properties(); properties.setProperty(IntegrationTestHelper.Config.ZNODE_PATH, "/storage/cli/test"); properties.setProperty(IntegrationTestHelper.Config.NUM_PARTITIONS, "1"); properties.setProperty(IntegrationTestHelper.Config.ZK_SESSION_TIMEOUT, "30000"); IntegrationTestHelper helper = new IntegrationTestHelper(properties); ReplicaId replicaId = new ReplicaId(0, String.format(TestReplicaSessionManager.CONNECT_STRING_TEMPLATE, 0)); ReplicaConnectionFactoryImpl connectionFactory = null; try { helper.startZooKeeperServer(); helper.startWaltzStorage(true); helper.setWaltzStorageAssignment(true); UUID key = helper.getClusterKey(); SslContext sslContext = Utils.getSslContext(helper.getSslConfigPath(), WaltzServerConfig.SERVER_SSL_CONFIG_PREFIX); ConnectionConfig config = TestUtils.makeConnectionConfig(1, key, sslContext); connectionFactory = new ReplicaConnectionFactoryImpl(helper.getStorageConnectString(), config); MockStoreSession mockStoreSession = new MockStoreSession(0, -1L, -1L); ReplicaSession replicaSession = new ReplicaSession(replicaId, 2, config, connectionFactory); RecoveryManager recoveryManager = TestUtils.mockRecoveryManager(1, -1L); replicaSession.open(recoveryManager, mockStoreSession); recoveryManager.highWaterMark(); ReplicaReader reader = replicaSession.reader; ReplicaWriter writer = replicaSession.writer; assertEquals(-1L, maxTransactionId(reader)); ArrayList<StoreAppendRequest> requests; synchronized (writer) { mockStoreSession.insertTransactions(0L, 10L); mockStoreSession.setHighWaterMark(9L); requests = TestUtils.makeStoreAppendRequests(0L, 10L); replicaSession.append(0L, requests, new Voting(1, 1)); assertEquals(-1L, maxTransactionId(reader)); mockStoreSession.insertTransactions(10L, 20L); mockStoreSession.setHighWaterMark(19L); requests = TestUtils.makeStoreAppendRequests(10L, 20L); replicaSession.append(10L, requests, new Voting(1, 1)); assertEquals(-1L, maxTransactionId(reader)); } int attempts = 10; while (maxTransactionId(reader) < 19L && attempts > 0) { Thread.sleep(20); attempts--; } assertEquals(19L, maxTransactionId(reader)); synchronized (writer) { mockStoreSession.insertTransactions(20L, 30L); mockStoreSession.setHighWaterMark(29L); requests = TestUtils.makeStoreAppendRequests(20L, 30L); replicaSession.append(20L, requests, new Voting(1, 1)); assertEquals(19L, maxTransactionId(reader)); } mockStoreSession.insertTransactions(30L, 40L); mockStoreSession.setHighWaterMark(39L); mockStoreSession.insertTransactions(40L, 50L); mockStoreSession.setHighWaterMark(49L); requests = TestUtils.makeStoreAppendRequests(30L, 40L); replicaSession.append(30L, requests, new Voting(1, 1)); requests = TestUtils.makeStoreAppendRequests(40L, 50L); replicaSession.append(40L, requests, new Voting(1, 1)); attempts = 10; while (maxTransactionId(reader) < 49L && attempts > 0) { Thread.sleep(20); attempts--; } assertEquals(49L, maxTransactionId(reader)); replicaSession.close(); mockStoreSession.close(); } finally { if (connectionFactory != null) { connectionFactory.close(); } helper.closeAll(); } }
private void catchUp(long nextTransactionId) throws Exception { long numRecords; synchronized (this) { numRecords = currentAppendRequestTransactionId - nextTransactionId; } while (numRecords > 0 && task.isRunning()) { int batchSize = numRecords < CATCH_UP_BATCH_SIZE ? (int) numRecords : CATCH_UP_BATCH_SIZE; ArrayList<Record> batch = storeSession.getRecordList(nextTransactionId, batchSize); writer.append(batch); nextTransactionId += batch.size(); synchronized (this) { numRecords = currentAppendRequestTransactionId - nextTransactionId; } } }
ReplicaSession extends LatencyWeightedRoute { private void catchUp(long nextTransactionId) throws Exception { long numRecords; synchronized (this) { numRecords = currentAppendRequestTransactionId - nextTransactionId; } while (numRecords > 0 && task.isRunning()) { int batchSize = numRecords < CATCH_UP_BATCH_SIZE ? (int) numRecords : CATCH_UP_BATCH_SIZE; ArrayList<Record> batch = storeSession.getRecordList(nextTransactionId, batchSize); writer.append(batch); nextTransactionId += batch.size(); synchronized (this) { numRecords = currentAppendRequestTransactionId - nextTransactionId; } } } }
ReplicaSession extends LatencyWeightedRoute { private void catchUp(long nextTransactionId) throws Exception { long numRecords; synchronized (this) { numRecords = currentAppendRequestTransactionId - nextTransactionId; } while (numRecords > 0 && task.isRunning()) { int batchSize = numRecords < CATCH_UP_BATCH_SIZE ? (int) numRecords : CATCH_UP_BATCH_SIZE; ArrayList<Record> batch = storeSession.getRecordList(nextTransactionId, batchSize); writer.append(batch); nextTransactionId += batch.size(); synchronized (this) { numRecords = currentAppendRequestTransactionId - nextTransactionId; } } } ReplicaSession(final ReplicaId replicaId, long sessionId, final ConnectionConfig config, final ReplicaConnectionFactory connectionFactory); }
ReplicaSession extends LatencyWeightedRoute { private void catchUp(long nextTransactionId) throws Exception { long numRecords; synchronized (this) { numRecords = currentAppendRequestTransactionId - nextTransactionId; } while (numRecords > 0 && task.isRunning()) { int batchSize = numRecords < CATCH_UP_BATCH_SIZE ? (int) numRecords : CATCH_UP_BATCH_SIZE; ArrayList<Record> batch = storeSession.getRecordList(nextTransactionId, batchSize); writer.append(batch); nextTransactionId += batch.size(); synchronized (this) { numRecords = currentAppendRequestTransactionId - nextTransactionId; } } } ReplicaSession(final ReplicaId replicaId, long sessionId, final ConnectionConfig config, final ReplicaConnectionFactory connectionFactory); void open(final RecoveryManager recoveryManager, final StoreSession storeSession); void close(); boolean isConnected(); @Override boolean isClosed(); void append(final long transactionId, final Iterable<StoreAppendRequest> requests, final Voting voting); long nextTransactionId(); Record getRecord(long transactionId); ArrayList<Record> getRecordList(long transactionId, int maxNumRecords); RecordHeader getRecordHeader(long transactionId); ArrayList<RecordHeader> getRecordHeaderList(long transactionId, int maxNumRecords); void awaitRecovery(); }
ReplicaSession extends LatencyWeightedRoute { private void catchUp(long nextTransactionId) throws Exception { long numRecords; synchronized (this) { numRecords = currentAppendRequestTransactionId - nextTransactionId; } while (numRecords > 0 && task.isRunning()) { int batchSize = numRecords < CATCH_UP_BATCH_SIZE ? (int) numRecords : CATCH_UP_BATCH_SIZE; ArrayList<Record> batch = storeSession.getRecordList(nextTransactionId, batchSize); writer.append(batch); nextTransactionId += batch.size(); synchronized (this) { numRecords = currentAppendRequestTransactionId - nextTransactionId; } } } ReplicaSession(final ReplicaId replicaId, long sessionId, final ConnectionConfig config, final ReplicaConnectionFactory connectionFactory); void open(final RecoveryManager recoveryManager, final StoreSession storeSession); void close(); boolean isConnected(); @Override boolean isClosed(); void append(final long transactionId, final Iterable<StoreAppendRequest> requests, final Voting voting); long nextTransactionId(); Record getRecord(long transactionId); ArrayList<Record> getRecordList(long transactionId, int maxNumRecords); RecordHeader getRecordHeader(long transactionId); ArrayList<RecordHeader> getRecordHeaderList(long transactionId, int maxNumRecords); void awaitRecovery(); final ReplicaId replicaId; }