@Test public void onReplicaEventTest() { TestHardwareLayout testHardwareLayout = new TestHardwareLayout("Alpha"); TestPartitionLayout testPartitionLayout = new TestPartitionLayout(testHardwareLayout, null); ClusterMap clusterMapManager = (new StaticClusterAgentsFactory(clusterMapConfig, testPartitionLayout.getPartitionLayout())).getClusterMap(); mockServerEventsAndVerify(clusterMapManager, clusterMapConfig, new ResourceState[]{ResourceState.Node_Up, ResourceState.Disk_Down, ResourceState.Replica_Up}, ServerErrorCode.Replica_Unavailable, new ResourceState[]{ResourceState.Node_Up, ResourceState.Disk_Up, ResourceState.Replica_Down}); mockServerEventsAndVerify(clusterMapManager, clusterMapConfig, new ResourceState[]{ResourceState.Node_Up, ResourceState.Disk_Down, ResourceState.Replica_Up}, ServerErrorCode.Temporarily_Disabled, new ResourceState[]{ResourceState.Node_Up, ResourceState.Disk_Up, ResourceState.Replica_Down}); mockServerEventsAndVerify(clusterMapManager, clusterMapConfig, new ResourceState[]{ResourceState.Node_Up, ResourceState.Disk_Down, ResourceState.Replica_Down}, ServerErrorCode.Replica_Unavailable, new ResourceState[]{ResourceState.Node_Up, ResourceState.Disk_Up, ResourceState.Replica_Down}); mockServerEventsAndVerify(clusterMapManager, clusterMapConfig, new ResourceState[]{ResourceState.Node_Up, ResourceState.Disk_Down, ResourceState.Replica_Down}, ServerErrorCode.Temporarily_Disabled, new ResourceState[]{ResourceState.Node_Up, ResourceState.Disk_Up, ResourceState.Replica_Down}); mockServerEventsAndVerify(clusterMapManager, clusterMapConfig, new ResourceState[]{ResourceState.Node_Up, ResourceState.Disk_Down, ResourceState.Replica_Down}, ServerErrorCode.Partition_ReadOnly, new ResourceState[]{ResourceState.Node_Up, ResourceState.Disk_Up, ResourceState.Replica_Up}); mockServerEventsAndVerify(clusterMapManager, clusterMapConfig, new ResourceState[]{ResourceState.Node_Up, ResourceState.Disk_Up, ResourceState.Replica_Up}, ServerErrorCode.IO_Error, new ResourceState[]{ResourceState.Node_Up, ResourceState.Disk_Down, ResourceState.Replica_Down}); mockServerEventsAndVerify(clusterMapManager, clusterMapConfig, new ResourceState[]{ResourceState.Node_Up, ResourceState.Disk_Up, ResourceState.Replica_Up}, ServerErrorCode.Disk_Unavailable, new ResourceState[]{ResourceState.Node_Up, ResourceState.Disk_Down, ResourceState.Replica_Down}); }
@Override public void onReplicaEvent(ReplicaId replicaId, ReplicaEventType event) { if (replicaId instanceof Replica) { switch (event) { case Disk_Error: if (replicaId.getReplicaType() == ReplicaType.DISK_BACKED) { ((Disk) replicaId.getDiskId()).onDiskError(); } break; case Disk_Ok: if (replicaId.getReplicaType() == ReplicaType.DISK_BACKED) { ((Disk) replicaId.getDiskId()).onDiskOk(); } break; case Node_Timeout: ((DataNode) replicaId.getDataNodeId()).onNodeTimeout(); break; case Node_Response: ((DataNode) replicaId.getDataNodeId()).onNodeResponse(); break; case Partition_ReadOnly: ((Partition) replicaId.getPartitionId()).onPartitionReadOnly(); break; case Replica_Unavailable: ((Replica) replicaId).onReplicaUnavailable(); break; case Replica_Available: ((Replica) replicaId).onReplicaResponse(); break; } } }
StaticClusterManager implements ClusterMap { @Override public void onReplicaEvent(ReplicaId replicaId, ReplicaEventType event) { if (replicaId instanceof Replica) { switch (event) { case Disk_Error: if (replicaId.getReplicaType() == ReplicaType.DISK_BACKED) { ((Disk) replicaId.getDiskId()).onDiskError(); } break; case Disk_Ok: if (replicaId.getReplicaType() == ReplicaType.DISK_BACKED) { ((Disk) replicaId.getDiskId()).onDiskOk(); } break; case Node_Timeout: ((DataNode) replicaId.getDataNodeId()).onNodeTimeout(); break; case Node_Response: ((DataNode) replicaId.getDataNodeId()).onNodeResponse(); break; case Partition_ReadOnly: ((Partition) replicaId.getPartitionId()).onPartitionReadOnly(); break; case Replica_Unavailable: ((Replica) replicaId).onReplicaUnavailable(); break; case Replica_Available: ((Replica) replicaId).onReplicaResponse(); break; } } } }
StaticClusterManager implements ClusterMap { @Override public void onReplicaEvent(ReplicaId replicaId, ReplicaEventType event) { if (replicaId instanceof Replica) { switch (event) { case Disk_Error: if (replicaId.getReplicaType() == ReplicaType.DISK_BACKED) { ((Disk) replicaId.getDiskId()).onDiskError(); } break; case Disk_Ok: if (replicaId.getReplicaType() == ReplicaType.DISK_BACKED) { ((Disk) replicaId.getDiskId()).onDiskOk(); } break; case Node_Timeout: ((DataNode) replicaId.getDataNodeId()).onNodeTimeout(); break; case Node_Response: ((DataNode) replicaId.getDataNodeId()).onNodeResponse(); break; case Partition_ReadOnly: ((Partition) replicaId.getPartitionId()).onPartitionReadOnly(); break; case Replica_Unavailable: ((Replica) replicaId).onReplicaUnavailable(); break; case Replica_Available: ((Replica) replicaId).onReplicaResponse(); break; } } } StaticClusterManager(PartitionLayout partitionLayout, String localDatacenterName, MetricRegistry metricRegistry); }
StaticClusterManager implements ClusterMap { @Override public void onReplicaEvent(ReplicaId replicaId, ReplicaEventType event) { if (replicaId instanceof Replica) { switch (event) { case Disk_Error: if (replicaId.getReplicaType() == ReplicaType.DISK_BACKED) { ((Disk) replicaId.getDiskId()).onDiskError(); } break; case Disk_Ok: if (replicaId.getReplicaType() == ReplicaType.DISK_BACKED) { ((Disk) replicaId.getDiskId()).onDiskOk(); } break; case Node_Timeout: ((DataNode) replicaId.getDataNodeId()).onNodeTimeout(); break; case Node_Response: ((DataNode) replicaId.getDataNodeId()).onNodeResponse(); break; case Partition_ReadOnly: ((Partition) replicaId.getPartitionId()).onPartitionReadOnly(); break; case Replica_Unavailable: ((Replica) replicaId).onReplicaUnavailable(); break; case Replica_Available: ((Replica) replicaId).onReplicaResponse(); break; } } } StaticClusterManager(PartitionLayout partitionLayout, String localDatacenterName, MetricRegistry metricRegistry); @Override List<PartitionId> getWritablePartitionIds(String partitionClass); @Override PartitionId getRandomWritablePartition(String partitionClass, List<PartitionId> partitionsToExclude); @Override List<PartitionId> getAllPartitionIds(String partitionClass); @Override PartitionId getPartitionIdFromStream(InputStream stream); @Override boolean hasDatacenter(String datacenterName); @Override byte getLocalDatacenterId(); @Override String getDatacenterName(byte id); @Override DataNodeId getDataNodeId(String hostname, int port); @Override List<ReplicaId> getReplicaIds(DataNodeId dataNodeId); @Override List<DataNodeId> getDataNodeIds(); @Override MetricRegistry getMetricRegistry(); @Override void registerClusterMapListener(ClusterMapChangeListener clusterMapChangeListener); @Override boolean equals(Object o); @Override void onReplicaEvent(ReplicaId replicaId, ReplicaEventType event); @Override JSONObject getSnapshot(); @Override ReplicaId getBootstrapReplica(String partitionIdStr, DataNodeId dataNodeId); @Override void close(); }
StaticClusterManager implements ClusterMap { @Override public void onReplicaEvent(ReplicaId replicaId, ReplicaEventType event) { if (replicaId instanceof Replica) { switch (event) { case Disk_Error: if (replicaId.getReplicaType() == ReplicaType.DISK_BACKED) { ((Disk) replicaId.getDiskId()).onDiskError(); } break; case Disk_Ok: if (replicaId.getReplicaType() == ReplicaType.DISK_BACKED) { ((Disk) replicaId.getDiskId()).onDiskOk(); } break; case Node_Timeout: ((DataNode) replicaId.getDataNodeId()).onNodeTimeout(); break; case Node_Response: ((DataNode) replicaId.getDataNodeId()).onNodeResponse(); break; case Partition_ReadOnly: ((Partition) replicaId.getPartitionId()).onPartitionReadOnly(); break; case Replica_Unavailable: ((Replica) replicaId).onReplicaUnavailable(); break; case Replica_Available: ((Replica) replicaId).onReplicaResponse(); break; } } } StaticClusterManager(PartitionLayout partitionLayout, String localDatacenterName, MetricRegistry metricRegistry); @Override List<PartitionId> getWritablePartitionIds(String partitionClass); @Override PartitionId getRandomWritablePartition(String partitionClass, List<PartitionId> partitionsToExclude); @Override List<PartitionId> getAllPartitionIds(String partitionClass); @Override PartitionId getPartitionIdFromStream(InputStream stream); @Override boolean hasDatacenter(String datacenterName); @Override byte getLocalDatacenterId(); @Override String getDatacenterName(byte id); @Override DataNodeId getDataNodeId(String hostname, int port); @Override List<ReplicaId> getReplicaIds(DataNodeId dataNodeId); @Override List<DataNodeId> getDataNodeIds(); @Override MetricRegistry getMetricRegistry(); @Override void registerClusterMapListener(ClusterMapChangeListener clusterMapChangeListener); @Override boolean equals(Object o); @Override void onReplicaEvent(ReplicaId replicaId, ReplicaEventType event); @Override JSONObject getSnapshot(); @Override ReplicaId getBootstrapReplica(String partitionIdStr, DataNodeId dataNodeId); @Override void close();  }