@Test public void testRecordList() throws Exception { int partitionId = new Random().nextInt(NUM_PARTITION); RecordListRequest msg = new RecordListRequest(-1, partitionId, -1, -1); adminServerHandler.process(msg); Mockito.verify(adminServerHandler).sendMessage(msgArg.capture(), boolArg.capture()); assertTrue(msgArg.getValue().exception.toString().contains("Partition:" + partitionId + " is not assigned.")); }
@Override protected void process(Message msg) throws Exception { AdminMessage message = (AdminMessage) msg; switch (message.type()) { case AdminMessageType.OPEN_REQUEST: try { storageManager.open(((AdminOpenRequest) message).key, ((AdminOpenRequest) message).numPartitions); success(); } catch (IllegalArgumentException ex) { failure(new StorageException("wrong key format")); } catch (StorageException ex) { failure(ex); } break; case AdminMessageType.PARTITION_AVAILABLE_REQUEST: try { storageManager.setPartitionAvailable(((PartitionAvailableRequest) msg).partitionId, ((PartitionAvailableRequest) msg).toggled); success(message); } catch (ConcurrentUpdateException | IOException | StorageException ex) { failure(ex, message); } break; case AdminMessageType.PARTITION_ASSIGNMENT_REQUEST: try { storageManager.setPartitionAssignment(((PartitionAssignmentRequest) msg).partitionId, ((PartitionAssignmentRequest) msg).toggled, ((PartitionAssignmentRequest) msg).deleteStorageFiles); success(message); } catch (ConcurrentUpdateException | IOException | StorageException ex) { failure(ex, message); } break; case AdminMessageType.RECORD_LIST_REQUEST: try { RecordListRequest recordListRequest = (RecordListRequest) message; Partition partition = storageManager.getPartition(recordListRequest.partitionId); if (partition != null) { ArrayList<Record> records = partition.getRecords(recordListRequest.transactionId, recordListRequest.maxNumRecords); sendMessage(new RecordListResponse(message.seqNum, recordListRequest.partitionId, records), true); } else { failure(new Exception("Partition:" + recordListRequest.partitionId + " is not assigned."), message); } } catch (IOException | StorageException ex) { failure(ex, message); } break; case AdminMessageType.LAST_SESSION_INFO_REQUEST: try { LastSessionInfoRequest lastSessionInfoRequest = (LastSessionInfoRequest) message; Partition partition = storageManager.getPartition(lastSessionInfoRequest.partitionId); if (partition != null) { SessionInfo sessionInfo = partition.getLastSessionInfo(); sendMessage(new LastSessionInfoResponse(message.seqNum, lastSessionInfoRequest.partitionId, sessionInfo), true); } else { failure(new Exception("Partition:" + lastSessionInfoRequest.partitionId + " is not assigned."), message); } } catch (StorageException ex) { failure(ex, message); } break; case AdminMessageType.METRICS_REQUEST: ObjectWriter ow = new ObjectMapper().writer().withDefaultPrettyPrinter(); String metricsJson = ow.writeValueAsString(MetricRegistry.getInstance()); sendMessage(new MetricsResponse(message.seqNum, metricsJson), true); break; case AdminMessageType.ASSIGNED_PARTITION_STATUS_REQUEST: Map<Integer, Boolean> partitionStatusMap = new HashMap<>(); Set<Integer> partitionSet = storageManager.getAssignedPartitionIds(); for (Integer id : partitionSet) { Partition partition = storageManager.getPartition(id); PartitionInfoSnapshot partitionInfoSnapshot = partition.getPartitionInfoSnapshot(); if (partitionInfoSnapshot.isAssigned && partitionInfoSnapshot.isAvailable) { partitionStatusMap.put(partitionInfoSnapshot.partitionId, true); } else { partitionStatusMap.put(partitionInfoSnapshot.partitionId, false); } } sendMessage(new AssignedPartitionStatusResponse(message.seqNum, partitionStatusMap), true); break; default: throw new IllegalStateException("unknown message type: " + msg.type()); } }
AdminServerHandler extends MessageHandler { @Override protected void process(Message msg) throws Exception { AdminMessage message = (AdminMessage) msg; switch (message.type()) { case AdminMessageType.OPEN_REQUEST: try { storageManager.open(((AdminOpenRequest) message).key, ((AdminOpenRequest) message).numPartitions); success(); } catch (IllegalArgumentException ex) { failure(new StorageException("wrong key format")); } catch (StorageException ex) { failure(ex); } break; case AdminMessageType.PARTITION_AVAILABLE_REQUEST: try { storageManager.setPartitionAvailable(((PartitionAvailableRequest) msg).partitionId, ((PartitionAvailableRequest) msg).toggled); success(message); } catch (ConcurrentUpdateException | IOException | StorageException ex) { failure(ex, message); } break; case AdminMessageType.PARTITION_ASSIGNMENT_REQUEST: try { storageManager.setPartitionAssignment(((PartitionAssignmentRequest) msg).partitionId, ((PartitionAssignmentRequest) msg).toggled, ((PartitionAssignmentRequest) msg).deleteStorageFiles); success(message); } catch (ConcurrentUpdateException | IOException | StorageException ex) { failure(ex, message); } break; case AdminMessageType.RECORD_LIST_REQUEST: try { RecordListRequest recordListRequest = (RecordListRequest) message; Partition partition = storageManager.getPartition(recordListRequest.partitionId); if (partition != null) { ArrayList<Record> records = partition.getRecords(recordListRequest.transactionId, recordListRequest.maxNumRecords); sendMessage(new RecordListResponse(message.seqNum, recordListRequest.partitionId, records), true); } else { failure(new Exception("Partition:" + recordListRequest.partitionId + " is not assigned."), message); } } catch (IOException | StorageException ex) { failure(ex, message); } break; case AdminMessageType.LAST_SESSION_INFO_REQUEST: try { LastSessionInfoRequest lastSessionInfoRequest = (LastSessionInfoRequest) message; Partition partition = storageManager.getPartition(lastSessionInfoRequest.partitionId); if (partition != null) { SessionInfo sessionInfo = partition.getLastSessionInfo(); sendMessage(new LastSessionInfoResponse(message.seqNum, lastSessionInfoRequest.partitionId, sessionInfo), true); } else { failure(new Exception("Partition:" + lastSessionInfoRequest.partitionId + " is not assigned."), message); } } catch (StorageException ex) { failure(ex, message); } break; case AdminMessageType.METRICS_REQUEST: ObjectWriter ow = new ObjectMapper().writer().withDefaultPrettyPrinter(); String metricsJson = ow.writeValueAsString(MetricRegistry.getInstance()); sendMessage(new MetricsResponse(message.seqNum, metricsJson), true); break; case AdminMessageType.ASSIGNED_PARTITION_STATUS_REQUEST: Map<Integer, Boolean> partitionStatusMap = new HashMap<>(); Set<Integer> partitionSet = storageManager.getAssignedPartitionIds(); for (Integer id : partitionSet) { Partition partition = storageManager.getPartition(id); PartitionInfoSnapshot partitionInfoSnapshot = partition.getPartitionInfoSnapshot(); if (partitionInfoSnapshot.isAssigned && partitionInfoSnapshot.isAvailable) { partitionStatusMap.put(partitionInfoSnapshot.partitionId, true); } else { partitionStatusMap.put(partitionInfoSnapshot.partitionId, false); } } sendMessage(new AssignedPartitionStatusResponse(message.seqNum, partitionStatusMap), true); break; default: throw new IllegalStateException("unknown message type: " + msg.type()); } } }
AdminServerHandler extends MessageHandler { @Override protected void process(Message msg) throws Exception { AdminMessage message = (AdminMessage) msg; switch (message.type()) { case AdminMessageType.OPEN_REQUEST: try { storageManager.open(((AdminOpenRequest) message).key, ((AdminOpenRequest) message).numPartitions); success(); } catch (IllegalArgumentException ex) { failure(new StorageException("wrong key format")); } catch (StorageException ex) { failure(ex); } break; case AdminMessageType.PARTITION_AVAILABLE_REQUEST: try { storageManager.setPartitionAvailable(((PartitionAvailableRequest) msg).partitionId, ((PartitionAvailableRequest) msg).toggled); success(message); } catch (ConcurrentUpdateException | IOException | StorageException ex) { failure(ex, message); } break; case AdminMessageType.PARTITION_ASSIGNMENT_REQUEST: try { storageManager.setPartitionAssignment(((PartitionAssignmentRequest) msg).partitionId, ((PartitionAssignmentRequest) msg).toggled, ((PartitionAssignmentRequest) msg).deleteStorageFiles); success(message); } catch (ConcurrentUpdateException | IOException | StorageException ex) { failure(ex, message); } break; case AdminMessageType.RECORD_LIST_REQUEST: try { RecordListRequest recordListRequest = (RecordListRequest) message; Partition partition = storageManager.getPartition(recordListRequest.partitionId); if (partition != null) { ArrayList<Record> records = partition.getRecords(recordListRequest.transactionId, recordListRequest.maxNumRecords); sendMessage(new RecordListResponse(message.seqNum, recordListRequest.partitionId, records), true); } else { failure(new Exception("Partition:" + recordListRequest.partitionId + " is not assigned."), message); } } catch (IOException | StorageException ex) { failure(ex, message); } break; case AdminMessageType.LAST_SESSION_INFO_REQUEST: try { LastSessionInfoRequest lastSessionInfoRequest = (LastSessionInfoRequest) message; Partition partition = storageManager.getPartition(lastSessionInfoRequest.partitionId); if (partition != null) { SessionInfo sessionInfo = partition.getLastSessionInfo(); sendMessage(new LastSessionInfoResponse(message.seqNum, lastSessionInfoRequest.partitionId, sessionInfo), true); } else { failure(new Exception("Partition:" + lastSessionInfoRequest.partitionId + " is not assigned."), message); } } catch (StorageException ex) { failure(ex, message); } break; case AdminMessageType.METRICS_REQUEST: ObjectWriter ow = new ObjectMapper().writer().withDefaultPrettyPrinter(); String metricsJson = ow.writeValueAsString(MetricRegistry.getInstance()); sendMessage(new MetricsResponse(message.seqNum, metricsJson), true); break; case AdminMessageType.ASSIGNED_PARTITION_STATUS_REQUEST: Map<Integer, Boolean> partitionStatusMap = new HashMap<>(); Set<Integer> partitionSet = storageManager.getAssignedPartitionIds(); for (Integer id : partitionSet) { Partition partition = storageManager.getPartition(id); PartitionInfoSnapshot partitionInfoSnapshot = partition.getPartitionInfoSnapshot(); if (partitionInfoSnapshot.isAssigned && partitionInfoSnapshot.isAvailable) { partitionStatusMap.put(partitionInfoSnapshot.partitionId, true); } else { partitionStatusMap.put(partitionInfoSnapshot.partitionId, false); } } sendMessage(new AssignedPartitionStatusResponse(message.seqNum, partitionStatusMap), true); break; default: throw new IllegalStateException("unknown message type: " + msg.type()); } } AdminServerHandler(StorageManager storageManager); }
AdminServerHandler extends MessageHandler { @Override protected void process(Message msg) throws Exception { AdminMessage message = (AdminMessage) msg; switch (message.type()) { case AdminMessageType.OPEN_REQUEST: try { storageManager.open(((AdminOpenRequest) message).key, ((AdminOpenRequest) message).numPartitions); success(); } catch (IllegalArgumentException ex) { failure(new StorageException("wrong key format")); } catch (StorageException ex) { failure(ex); } break; case AdminMessageType.PARTITION_AVAILABLE_REQUEST: try { storageManager.setPartitionAvailable(((PartitionAvailableRequest) msg).partitionId, ((PartitionAvailableRequest) msg).toggled); success(message); } catch (ConcurrentUpdateException | IOException | StorageException ex) { failure(ex, message); } break; case AdminMessageType.PARTITION_ASSIGNMENT_REQUEST: try { storageManager.setPartitionAssignment(((PartitionAssignmentRequest) msg).partitionId, ((PartitionAssignmentRequest) msg).toggled, ((PartitionAssignmentRequest) msg).deleteStorageFiles); success(message); } catch (ConcurrentUpdateException | IOException | StorageException ex) { failure(ex, message); } break; case AdminMessageType.RECORD_LIST_REQUEST: try { RecordListRequest recordListRequest = (RecordListRequest) message; Partition partition = storageManager.getPartition(recordListRequest.partitionId); if (partition != null) { ArrayList<Record> records = partition.getRecords(recordListRequest.transactionId, recordListRequest.maxNumRecords); sendMessage(new RecordListResponse(message.seqNum, recordListRequest.partitionId, records), true); } else { failure(new Exception("Partition:" + recordListRequest.partitionId + " is not assigned."), message); } } catch (IOException | StorageException ex) { failure(ex, message); } break; case AdminMessageType.LAST_SESSION_INFO_REQUEST: try { LastSessionInfoRequest lastSessionInfoRequest = (LastSessionInfoRequest) message; Partition partition = storageManager.getPartition(lastSessionInfoRequest.partitionId); if (partition != null) { SessionInfo sessionInfo = partition.getLastSessionInfo(); sendMessage(new LastSessionInfoResponse(message.seqNum, lastSessionInfoRequest.partitionId, sessionInfo), true); } else { failure(new Exception("Partition:" + lastSessionInfoRequest.partitionId + " is not assigned."), message); } } catch (StorageException ex) { failure(ex, message); } break; case AdminMessageType.METRICS_REQUEST: ObjectWriter ow = new ObjectMapper().writer().withDefaultPrettyPrinter(); String metricsJson = ow.writeValueAsString(MetricRegistry.getInstance()); sendMessage(new MetricsResponse(message.seqNum, metricsJson), true); break; case AdminMessageType.ASSIGNED_PARTITION_STATUS_REQUEST: Map<Integer, Boolean> partitionStatusMap = new HashMap<>(); Set<Integer> partitionSet = storageManager.getAssignedPartitionIds(); for (Integer id : partitionSet) { Partition partition = storageManager.getPartition(id); PartitionInfoSnapshot partitionInfoSnapshot = partition.getPartitionInfoSnapshot(); if (partitionInfoSnapshot.isAssigned && partitionInfoSnapshot.isAvailable) { partitionStatusMap.put(partitionInfoSnapshot.partitionId, true); } else { partitionStatusMap.put(partitionInfoSnapshot.partitionId, false); } } sendMessage(new AssignedPartitionStatusResponse(message.seqNum, partitionStatusMap), true); break; default: throw new IllegalStateException("unknown message type: " + msg.type()); } } AdminServerHandler(StorageManager storageManager);  }
AdminServerHandler extends MessageHandler { @Override protected void process(Message msg) throws Exception { AdminMessage message = (AdminMessage) msg; switch (message.type()) { case AdminMessageType.OPEN_REQUEST: try { storageManager.open(((AdminOpenRequest) message).key, ((AdminOpenRequest) message).numPartitions); success(); } catch (IllegalArgumentException ex) { failure(new StorageException("wrong key format")); } catch (StorageException ex) { failure(ex); } break; case AdminMessageType.PARTITION_AVAILABLE_REQUEST: try { storageManager.setPartitionAvailable(((PartitionAvailableRequest) msg).partitionId, ((PartitionAvailableRequest) msg).toggled); success(message); } catch (ConcurrentUpdateException | IOException | StorageException ex) { failure(ex, message); } break; case AdminMessageType.PARTITION_ASSIGNMENT_REQUEST: try { storageManager.setPartitionAssignment(((PartitionAssignmentRequest) msg).partitionId, ((PartitionAssignmentRequest) msg).toggled, ((PartitionAssignmentRequest) msg).deleteStorageFiles); success(message); } catch (ConcurrentUpdateException | IOException | StorageException ex) { failure(ex, message); } break; case AdminMessageType.RECORD_LIST_REQUEST: try { RecordListRequest recordListRequest = (RecordListRequest) message; Partition partition = storageManager.getPartition(recordListRequest.partitionId); if (partition != null) { ArrayList<Record> records = partition.getRecords(recordListRequest.transactionId, recordListRequest.maxNumRecords); sendMessage(new RecordListResponse(message.seqNum, recordListRequest.partitionId, records), true); } else { failure(new Exception("Partition:" + recordListRequest.partitionId + " is not assigned."), message); } } catch (IOException | StorageException ex) { failure(ex, message); } break; case AdminMessageType.LAST_SESSION_INFO_REQUEST: try { LastSessionInfoRequest lastSessionInfoRequest = (LastSessionInfoRequest) message; Partition partition = storageManager.getPartition(lastSessionInfoRequest.partitionId); if (partition != null) { SessionInfo sessionInfo = partition.getLastSessionInfo(); sendMessage(new LastSessionInfoResponse(message.seqNum, lastSessionInfoRequest.partitionId, sessionInfo), true); } else { failure(new Exception("Partition:" + lastSessionInfoRequest.partitionId + " is not assigned."), message); } } catch (StorageException ex) { failure(ex, message); } break; case AdminMessageType.METRICS_REQUEST: ObjectWriter ow = new ObjectMapper().writer().withDefaultPrettyPrinter(); String metricsJson = ow.writeValueAsString(MetricRegistry.getInstance()); sendMessage(new MetricsResponse(message.seqNum, metricsJson), true); break; case AdminMessageType.ASSIGNED_PARTITION_STATUS_REQUEST: Map<Integer, Boolean> partitionStatusMap = new HashMap<>(); Set<Integer> partitionSet = storageManager.getAssignedPartitionIds(); for (Integer id : partitionSet) { Partition partition = storageManager.getPartition(id); PartitionInfoSnapshot partitionInfoSnapshot = partition.getPartitionInfoSnapshot(); if (partitionInfoSnapshot.isAssigned && partitionInfoSnapshot.isAvailable) { partitionStatusMap.put(partitionInfoSnapshot.partitionId, true); } else { partitionStatusMap.put(partitionInfoSnapshot.partitionId, false); } } sendMessage(new AssignedPartitionStatusResponse(message.seqNum, partitionStatusMap), true); break; default: throw new IllegalStateException("unknown message type: " + msg.type()); } } AdminServerHandler(StorageManager storageManager);   }