@Test public void cloudReplicaAdditionTest() throws Exception { StorageManager storageManager = new StorageManager(storeConfig, new DiskManagerConfig(verifiableProperties), Utils.newScheduler(1, true), clusterMap.getMetricRegistry(), null, clusterMap, currentNode, null, Collections.singletonList(mockHelixParticipant), new MockTime(), null, new InMemAccountService(false, false)); CloudToStoreReplicationManager cloudToStoreReplicationManager = new CloudToStoreReplicationManager(replicationConfig, clusterMapConfig, storeConfig, storageManager, storeKeyFactory, clusterMap, mockScheduler, currentNode, null, clusterMap.getMetricRegistry(), null, storeKeyConverterFactory, serverConfig.serverMessageTransformer, mockClusterSpectator, mockHelixParticipant); storageManager.start(); cloudToStoreReplicationManager.start(); mockClusterSpectator.spectate(); mockHelixParticipant.onPartitionBecomeLeaderFromStandby(NEW_PARTITION_NAME); assertNull("Cloud replica thread should not be created", TestUtils.getThreadByThisName(VCR_REPLICA_THREAD_PREFIX)); PartitionId newPartition = new MockPartitionId(Long.parseLong(NEW_PARTITION_NAME), MockClusterMap.DEFAULT_PARTITION_CLASS, clusterMap.getDataNodes(), 0); ReplicaId replicaToAdd = newPartition.getReplicaIds().get(0); assertTrue("Adding new store should succeed", storageManager.addBlobStore(replicaToAdd)); storageManager.shutdownBlobStore(newPartition); mockHelixParticipant.onPartitionBecomeLeaderFromStandby(NEW_PARTITION_NAME); assertNull("Cloud replica thread should not be created", TestUtils.getThreadByThisName(VCR_REPLICA_THREAD_PREFIX)); storageManager.startBlobStore(newPartition); mockHelixParticipant.onPartitionBecomeLeaderFromStandby(NEW_PARTITION_NAME); assertNotNull("Cloud replica thread should be created for DC1", TestUtils.getThreadByThisName(VCR_REPLICA_THREAD_PREFIX)); cloudToStoreReplicationManager.shutdown(); storageManager.shutdown(); }
@Override public void start() { vcrClusterSpectator.registerInstanceConfigChangeListener(new InstanceConfigChangeListenerImpl()); vcrClusterSpectator.registerLiveInstanceChangeListener(new LiveInstanceChangeListenerImpl()); clusterParticipant.registerPartitionStateChangeListener( StateModelListenerType.CloudToStoreReplicationManagerListener, new PartitionStateChangeListenerImpl()); scheduler.scheduleAtFixedRate(persistor, replicationConfig.replicationTokenFlushDelaySeconds, replicationConfig.replicationTokenFlushIntervalSeconds, TimeUnit.SECONDS); }
CloudToStoreReplicationManager extends ReplicationEngine { @Override public void start() { vcrClusterSpectator.registerInstanceConfigChangeListener(new InstanceConfigChangeListenerImpl()); vcrClusterSpectator.registerLiveInstanceChangeListener(new LiveInstanceChangeListenerImpl()); clusterParticipant.registerPartitionStateChangeListener( StateModelListenerType.CloudToStoreReplicationManagerListener, new PartitionStateChangeListenerImpl()); scheduler.scheduleAtFixedRate(persistor, replicationConfig.replicationTokenFlushDelaySeconds, replicationConfig.replicationTokenFlushIntervalSeconds, TimeUnit.SECONDS); } }
CloudToStoreReplicationManager extends ReplicationEngine { @Override public void start() { vcrClusterSpectator.registerInstanceConfigChangeListener(new InstanceConfigChangeListenerImpl()); vcrClusterSpectator.registerLiveInstanceChangeListener(new LiveInstanceChangeListenerImpl()); clusterParticipant.registerPartitionStateChangeListener( StateModelListenerType.CloudToStoreReplicationManagerListener, new PartitionStateChangeListenerImpl()); scheduler.scheduleAtFixedRate(persistor, replicationConfig.replicationTokenFlushDelaySeconds, replicationConfig.replicationTokenFlushIntervalSeconds, TimeUnit.SECONDS); } CloudToStoreReplicationManager(ReplicationConfig replicationConfig, ClusterMapConfig clusterMapConfig,
      StoreConfig storeConfig, StoreManager storeManager, StoreKeyFactory storeKeyFactory, ClusterMap clusterMap,
      ScheduledExecutorService scheduler, DataNodeId currentNode, ConnectionPool connectionPool,
      MetricRegistry metricRegistry, NotificationSystem requestNotification,
      StoreKeyConverterFactory storeKeyConverterFactory, String transformerClassName,
      ClusterSpectator vcrClusterSpectator, ClusterParticipant clusterParticipant); }
CloudToStoreReplicationManager extends ReplicationEngine { @Override public void start() { vcrClusterSpectator.registerInstanceConfigChangeListener(new InstanceConfigChangeListenerImpl()); vcrClusterSpectator.registerLiveInstanceChangeListener(new LiveInstanceChangeListenerImpl()); clusterParticipant.registerPartitionStateChangeListener( StateModelListenerType.CloudToStoreReplicationManagerListener, new PartitionStateChangeListenerImpl()); scheduler.scheduleAtFixedRate(persistor, replicationConfig.replicationTokenFlushDelaySeconds, replicationConfig.replicationTokenFlushIntervalSeconds, TimeUnit.SECONDS); } CloudToStoreReplicationManager(ReplicationConfig replicationConfig, ClusterMapConfig clusterMapConfig,
      StoreConfig storeConfig, StoreManager storeManager, StoreKeyFactory storeKeyFactory, ClusterMap clusterMap,
      ScheduledExecutorService scheduler, DataNodeId currentNode, ConnectionPool connectionPool,
      MetricRegistry metricRegistry, NotificationSystem requestNotification,
      StoreKeyConverterFactory storeKeyConverterFactory, String transformerClassName,
      ClusterSpectator vcrClusterSpectator, ClusterParticipant clusterParticipant); @Override void start(); }
CloudToStoreReplicationManager extends ReplicationEngine { @Override public void start() { vcrClusterSpectator.registerInstanceConfigChangeListener(new InstanceConfigChangeListenerImpl()); vcrClusterSpectator.registerLiveInstanceChangeListener(new LiveInstanceChangeListenerImpl()); clusterParticipant.registerPartitionStateChangeListener( StateModelListenerType.CloudToStoreReplicationManagerListener, new PartitionStateChangeListenerImpl()); scheduler.scheduleAtFixedRate(persistor, replicationConfig.replicationTokenFlushDelaySeconds, replicationConfig.replicationTokenFlushIntervalSeconds, TimeUnit.SECONDS); } CloudToStoreReplicationManager(ReplicationConfig replicationConfig, ClusterMapConfig clusterMapConfig,
      StoreConfig storeConfig, StoreManager storeManager, StoreKeyFactory storeKeyFactory, ClusterMap clusterMap,
      ScheduledExecutorService scheduler, DataNodeId currentNode, ConnectionPool connectionPool,
      MetricRegistry metricRegistry, NotificationSystem requestNotification,
      StoreKeyConverterFactory storeKeyConverterFactory, String transformerClassName,
      ClusterSpectator vcrClusterSpectator, ClusterParticipant clusterParticipant); @Override void start();  }