@Test public void cloudReplicaRemovalTest() 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(); PartitionId localPartition = storageManager.getLocalPartitions().iterator().next(); mockHelixParticipant.onPartitionBecomeLeaderFromStandby(localPartition.toPathString()); String replicaPath = Cloud_Replica_Keyword + File.separator + localPartition.toPathString() + File.separator + localPartition.toPathString(); RemoteReplicaInfo remoteReplicaInfo = cloudToStoreReplicationManager.getRemoteReplicaInfo(localPartition, vcrNode.getHostname(), replicaPath); assertNotNull("Remote replica info should not be null", remoteReplicaInfo); assertEquals("There should be only one cloud replica thread created", 1, TestUtils.getAllThreadsByThisName(VCR_REPLICA_THREAD_PREFIX).size()); mockHelixParticipant.onPartitionBecomeStandbyFromLeader(NEW_PARTITION_NAME); assertEquals("There should be only one cloud replica thread created", 1, TestUtils.getAllThreadsByThisName(VCR_REPLICA_THREAD_PREFIX).size()); mockHelixParticipant.onPartitionBecomeStandbyFromLeader(localPartition.toPathString()); assertNull("Cloud replica should be removed and no thread is assigned to it", remoteReplicaInfo.getReplicaThread()); 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();  }