@Test public void version0AndCurrentVersionRetrieveTest() throws Exception { DiskTokenPersistor diskTokenPersistor = new DiskTokenPersistor(REPLICA_TOKEN_FILENAME, mountPathToPartitionInfoList, new ReplicationMetrics(new MetricRegistry(), Collections.emptyList()), clusterMap, findTokenHelper, mockStoreManager); List<RemoteReplicaInfo.ReplicaTokenInfo> replicaTokenInfoList = mountPathToReplicaTokenInfos.get(replicaId.getMountPath()); persistVersion0(replicaId.getMountPath(), replicaTokenInfoList); List<RemoteReplicaInfo.ReplicaTokenInfo> retrievedReplicaTokenInfos = diskTokenPersistor.retrieve(replicaId.getMountPath()); assertEquals("Number of tokens doesn't match.", replicaTokenInfoList.size(), retrievedReplicaTokenInfos.size()); for (int i = 0; i < replicaTokenInfoList.size(); i++) { assertArrayEquals("Token is not correct.", replicaTokenInfoList.get(i).getReplicaToken().toBytes(), retrievedReplicaTokenInfos.get(i).getReplicaToken().toBytes()); } }
@Override public List<ReplicaTokenInfo> retrieve(String mountPath) throws ReplicationException { File replicaTokenFile = new File(mountPath, replicaTokenFileName); if (replicaTokenFile.exists()) { try (FileInputStream fileInputStream = new FileInputStream(replicaTokenFile)) { return replicaTokenSerde.deserializeTokens(fileInputStream); } catch (IOException e) { throw new ReplicationException("IO error while reading from replica token file at mount path " + mountPath, e); } } else { return Collections.emptyList(); } }
DiskTokenPersistor extends ReplicaTokenPersistor { @Override public List<ReplicaTokenInfo> retrieve(String mountPath) throws ReplicationException { File replicaTokenFile = new File(mountPath, replicaTokenFileName); if (replicaTokenFile.exists()) { try (FileInputStream fileInputStream = new FileInputStream(replicaTokenFile)) { return replicaTokenSerde.deserializeTokens(fileInputStream); } catch (IOException e) { throw new ReplicationException("IO error while reading from replica token file at mount path " + mountPath, e); } } else { return Collections.emptyList(); } } }
DiskTokenPersistor extends ReplicaTokenPersistor { @Override public List<ReplicaTokenInfo> retrieve(String mountPath) throws ReplicationException { File replicaTokenFile = new File(mountPath, replicaTokenFileName); if (replicaTokenFile.exists()) { try (FileInputStream fileInputStream = new FileInputStream(replicaTokenFile)) { return replicaTokenSerde.deserializeTokens(fileInputStream); } catch (IOException e) { throw new ReplicationException("IO error while reading from replica token file at mount path " + mountPath, e); } } else { return Collections.emptyList(); } } DiskTokenPersistor(String replicaTokenFileName, Map<String, Set<PartitionInfo>> partitionGroupedByMountPath,
      ReplicationMetrics replicationMetrics, ClusterMap clusterMap, FindTokenHelper tokenHelper,
      StoreManager storeManager); }
DiskTokenPersistor extends ReplicaTokenPersistor { @Override public List<ReplicaTokenInfo> retrieve(String mountPath) throws ReplicationException { File replicaTokenFile = new File(mountPath, replicaTokenFileName); if (replicaTokenFile.exists()) { try (FileInputStream fileInputStream = new FileInputStream(replicaTokenFile)) { return replicaTokenSerde.deserializeTokens(fileInputStream); } catch (IOException e) { throw new ReplicationException("IO error while reading from replica token file at mount path " + mountPath, e); } } else { return Collections.emptyList(); } } DiskTokenPersistor(String replicaTokenFileName, Map<String, Set<PartitionInfo>> partitionGroupedByMountPath,
      ReplicationMetrics replicationMetrics, ClusterMap clusterMap, FindTokenHelper tokenHelper,
      StoreManager storeManager); @Override List<ReplicaTokenInfo> retrieve(String mountPath); }
DiskTokenPersistor extends ReplicaTokenPersistor { @Override public List<ReplicaTokenInfo> retrieve(String mountPath) throws ReplicationException { File replicaTokenFile = new File(mountPath, replicaTokenFileName); if (replicaTokenFile.exists()) { try (FileInputStream fileInputStream = new FileInputStream(replicaTokenFile)) { return replicaTokenSerde.deserializeTokens(fileInputStream); } catch (IOException e) { throw new ReplicationException("IO error while reading from replica token file at mount path " + mountPath, e); } } else { return Collections.emptyList(); } } DiskTokenPersistor(String replicaTokenFileName, Map<String, Set<PartitionInfo>> partitionGroupedByMountPath,
      ReplicationMetrics replicationMetrics, ClusterMap clusterMap, FindTokenHelper tokenHelper,
      StoreManager storeManager); @Override List<ReplicaTokenInfo> retrieve(String mountPath);  }