From 5aca5c06ded7761e1af879da4bc2857a19458d50 Mon Sep 17 00:00:00 2001 From: Jay J Wylie Date: Wed, 13 Mar 2013 10:34:21 -0700 Subject: [PATCH] change maxRecords to recordsPerPartition in fetch API and protobuf These are cosmetic changes. The client-side and server-side code does not properly do recordsPerPartition yet. Added a few TODOs in the code too. --- .../client/protocol/admin/AdminClient.java | 20 +- .../client/protocol/pb/VAdminProto.java | 377 +++++++++--------- .../FetchEntriesStreamRequestHandler.java | 6 +- .../admin/FetchKeysStreamRequestHandler.java | 6 +- ...hPartitionEntriesStreamRequestHandler.java | 8 +- ...etchPartitionKeysStreamRequestHandler.java | 7 +- .../admin/FetchStreamRequestHandler.java | 13 +- src/java/voldemort/utils/Entropy.java | 3 + src/java/voldemort/utils/KeySamplerCLI.java | 44 +- src/proto/voldemort-admin.proto | 3 +- 10 files changed, 254 insertions(+), 233 deletions(-) diff --git a/src/java/voldemort/client/protocol/admin/AdminClient.java b/src/java/voldemort/client/protocol/admin/AdminClient.java index 69fe275b33..3857b2a540 100644 --- a/src/java/voldemort/client/protocol/admin/AdminClient.java +++ b/src/java/voldemort/client/protocol/admin/AdminClient.java @@ -1444,7 +1444,7 @@ private void initiateFetchRequest(DataOutputStream outputStream, boolean fetchValues, boolean fetchMasterEntries, Cluster initialCluster, - long maxRecords) throws IOException { + long recordsPerPartition) throws IOException { HashMap> filteredReplicaToPartitionList = Maps.newHashMap(); if(fetchMasterEntries) { if(!replicaToPartitionList.containsKey(0)) { @@ -1459,7 +1459,7 @@ private void initiateFetchRequest(DataOutputStream outputStream, .setFetchValues(fetchValues) .addAllReplicaToPartition(ProtoUtils.encodePartitionTuple(filteredReplicaToPartitionList)) .setStore(storeName) - .setMaxRecords(maxRecords); + .setRecordsPerPartition(recordsPerPartition); try { if(filter != null) { @@ -1582,14 +1582,14 @@ public Iterator>> fetchEntries(int nodeId, List partitionList, VoldemortFilter filter, boolean fetchMasterEntries, - long maxRecords) { + long recordsPerPartition) { return fetchEntries(nodeId, storeName, helperOps.getReplicaToPartitionMap(nodeId, storeName, partitionList), filter, fetchMasterEntries, null, - maxRecords); + recordsPerPartition); } /** @@ -1656,7 +1656,7 @@ public Iterator>> fetchEntries(int nodeId, VoldemortFilter filter, boolean fetchMasterEntries, Cluster initialCluster, - long maxRecords) { + long recordsPerPartition) { Node node = AdminClient.this.getAdminClientCluster().getNodeById(nodeId); final SocketDestination destination = new SocketDestination(node.getHost(), @@ -1674,7 +1674,7 @@ public Iterator>> fetchEntries(int nodeId, true, fetchMasterEntries, initialCluster, - maxRecords); + recordsPerPartition); } catch(IOException e) { helperOps.close(sands.getSocket()); socketPool.checkin(destination, sands); @@ -1800,14 +1800,14 @@ public Iterator fetchKeys(int nodeId, List partitionList, VoldemortFilter filter, boolean fetchMasterEntries, - long maxRecords) { + long recordsPerPartition) { return fetchKeys(nodeId, storeName, helperOps.getReplicaToPartitionMap(nodeId, storeName, partitionList), filter, fetchMasterEntries, null, - maxRecords); + recordsPerPartition); } /** @@ -1852,7 +1852,7 @@ public Iterator fetchKeys(int nodeId, VoldemortFilter filter, boolean fetchMasterEntries, Cluster initialCluster, - long maxRecords) { + long recordsPerPartition) { Node node = AdminClient.this.getAdminClientCluster().getNodeById(nodeId); final SocketDestination destination = new SocketDestination(node.getHost(), node.getAdminPort(), @@ -1869,7 +1869,7 @@ public Iterator fetchKeys(int nodeId, false, fetchMasterEntries, initialCluster, - maxRecords); + recordsPerPartition); } catch(IOException e) { helperOps.close(sands.getSocket()); socketPool.checkin(destination, sands); diff --git a/src/java/voldemort/client/protocol/pb/VAdminProto.java b/src/java/voldemort/client/protocol/pb/VAdminProto.java index 31b1e4b5e4..e34aec3563 100644 --- a/src/java/voldemort/client/protocol/pb/VAdminProto.java +++ b/src/java/voldemort/client/protocol/pb/VAdminProto.java @@ -4507,12 +4507,12 @@ public voldemort.client.protocol.pb.VAdminProto.PartitionTuple getReplicaToParti public boolean hasFetchOrphaned() { return hasFetchOrphaned; } public boolean getFetchOrphaned() { return fetchOrphaned_; } - // optional int64 max_records = 8; - public static final int MAX_RECORDS_FIELD_NUMBER = 8; - private boolean hasMaxRecords; - private long maxRecords_ = 0L; - public boolean hasMaxRecords() { return hasMaxRecords; } - public long getMaxRecords() { return maxRecords_; } + // optional int64 records_per_partition = 8; + public static final int RECORDS_PER_PARTITION_FIELD_NUMBER = 8; + private boolean hasRecordsPerPartition; + private long recordsPerPartition_ = 0L; + public boolean hasRecordsPerPartition() { return hasRecordsPerPartition; } + public long getRecordsPerPartition() { return recordsPerPartition_; } private void initFields() { filter_ = voldemort.client.protocol.pb.VAdminProto.VoldemortFilter.getDefaultInstance(); @@ -4552,8 +4552,8 @@ public void writeTo(com.google.protobuf.CodedOutputStream output) if (hasFetchOrphaned()) { output.writeBool(7, getFetchOrphaned()); } - if (hasMaxRecords()) { - output.writeInt64(8, getMaxRecords()); + if (hasRecordsPerPartition()) { + output.writeInt64(8, getRecordsPerPartition()); } getUnknownFields().writeTo(output); } @@ -4592,9 +4592,9 @@ public int getSerializedSize() { size += com.google.protobuf.CodedOutputStream .computeBoolSize(7, getFetchOrphaned()); } - if (hasMaxRecords()) { + if (hasRecordsPerPartition()) { size += com.google.protobuf.CodedOutputStream - .computeInt64Size(8, getMaxRecords()); + .computeInt64Size(8, getRecordsPerPartition()); } size += getUnknownFields().getSerializedSize(); memoizedSerializedSize = size; @@ -4782,8 +4782,8 @@ public Builder mergeFrom(voldemort.client.protocol.pb.VAdminProto.FetchPartition if (other.hasFetchOrphaned()) { setFetchOrphaned(other.getFetchOrphaned()); } - if (other.hasMaxRecords()) { - setMaxRecords(other.getMaxRecords()); + if (other.hasRecordsPerPartition()) { + setRecordsPerPartition(other.getRecordsPerPartition()); } this.mergeUnknownFields(other.getUnknownFields()); return this; @@ -4846,7 +4846,7 @@ public Builder mergeFrom( break; } case 64: { - setMaxRecords(input.readInt64()); + setRecordsPerPartition(input.readInt64()); break; } } @@ -5038,21 +5038,21 @@ public Builder clearFetchOrphaned() { return this; } - // optional int64 max_records = 8; - public boolean hasMaxRecords() { - return result.hasMaxRecords(); + // optional int64 records_per_partition = 8; + public boolean hasRecordsPerPartition() { + return result.hasRecordsPerPartition(); } - public long getMaxRecords() { - return result.getMaxRecords(); + public long getRecordsPerPartition() { + return result.getRecordsPerPartition(); } - public Builder setMaxRecords(long value) { - result.hasMaxRecords = true; - result.maxRecords_ = value; + public Builder setRecordsPerPartition(long value) { + result.hasRecordsPerPartition = true; + result.recordsPerPartition_ = value; return this; } - public Builder clearMaxRecords() { - result.hasMaxRecords = false; - result.maxRecords_ = 0L; + public Builder clearRecordsPerPartition() { + result.hasRecordsPerPartition = false; + result.recordsPerPartition_ = 0L; return this; } @@ -23163,175 +23163,176 @@ public Builder clearReserveMemory() { "e\022\037\n\005error\030\001 \001(\0132\020.voldemort.Error\"d\n\032Fe" + "tchPartitionFilesRequest\022\r\n\005store\030\001 \002(\t\022" + "7\n\024replica_to_partition\030\002 \003(\0132\031.voldemor" + - "t.PartitionTuple\"\232\002\n\034FetchPartitionEntri" + + "t.PartitionTuple\"\244\002\n\034FetchPartitionEntri" + "esRequest\0227\n\024replica_to_partition\030\001 \003(\0132" + "\031.voldemort.PartitionTuple\022\r\n\005store\030\002 \002(" + "\t\022*\n\006filter\030\003 \001(\0132\032.voldemort.VoldemortF", "ilter\022\024\n\014fetch_values\030\004 \001(\010\022*\n\"OBSOLETE_" + "_DO_NOT_USE__skip_records\030\005 \001(\003\022\027\n\017initi" + "al_cluster\030\006 \001(\t\022\026\n\016fetch_orphaned\030\007 \001(\010" + - "\022\023\n\013max_records\030\010 \001(\003\"\201\001\n\035FetchPartition" + - "EntriesResponse\0222\n\017partition_entry\030\001 \001(\013" + - "2\031.voldemort.PartitionEntry\022\013\n\003key\030\002 \001(\014" + - "\022\037\n\005error\030\003 \001(\0132\020.voldemort.Error\"\254\001\n\035De" + - "letePartitionEntriesRequest\022\r\n\005store\030\001 \002" + - "(\t\0227\n\024replica_to_partition\030\002 \003(\0132\031.volde" + - "mort.PartitionTuple\022*\n\006filter\030\003 \001(\0132\032.vo", - "ldemort.VoldemortFilter\022\027\n\017initial_clust" + - "er\030\004 \001(\t\"P\n\036DeletePartitionEntriesRespon" + - "se\022\r\n\005count\030\001 \001(\003\022\037\n\005error\030\002 \001(\0132\020.volde" + - "mort.Error\"\317\001\n\035InitiateFetchAndUpdateReq" + - "uest\022\017\n\007node_id\030\001 \002(\005\022\r\n\005store\030\002 \002(\t\022*\n\006" + - "filter\030\003 \001(\0132\032.voldemort.VoldemortFilter" + - "\0227\n\024replica_to_partition\030\004 \003(\0132\031.voldemo" + - "rt.PartitionTuple\022\027\n\017initial_cluster\030\005 \001" + - "(\t\022\020\n\010optimize\030\006 \001(\010\"1\n\033AsyncOperationSt" + - "atusRequest\022\022\n\nrequest_id\030\001 \002(\005\"/\n\031Async", - "OperationStopRequest\022\022\n\nrequest_id\030\001 \002(\005" + - "\"=\n\032AsyncOperationStopResponse\022\037\n\005error\030" + - "\001 \001(\0132\020.voldemort.Error\"2\n\031AsyncOperatio" + - "nListRequest\022\025\n\rshow_complete\030\002 \002(\010\"R\n\032A" + - "syncOperationListResponse\022\023\n\013request_ids" + - "\030\001 \003(\005\022\037\n\005error\030\002 \001(\0132\020.voldemort.Error\"" + - ":\n\016PartitionTuple\022\024\n\014replica_type\030\001 \002(\005\022" + - "\022\n\npartitions\030\002 \003(\005\"e\n\026PerStorePartition" + - "Tuple\022\022\n\nstore_name\030\001 \002(\t\0227\n\024replica_to_" + - "partition\030\002 \003(\0132\031.voldemort.PartitionTup", - "le\"\370\001\n\031RebalancePartitionInfoMap\022\022\n\nstea" + - "ler_id\030\001 \002(\005\022\020\n\010donor_id\030\002 \002(\005\022\017\n\007attemp" + - "t\030\003 \002(\005\022C\n\030replica_to_add_partition\030\004 \003(" + - "\0132!.voldemort.PerStorePartitionTuple\022F\n\033" + - "replica_to_delete_partition\030\005 \003(\0132!.vold" + - "emort.PerStorePartitionTuple\022\027\n\017initial_" + - "cluster\030\006 \002(\t\"f\n\034InitiateRebalanceNodeRe" + - "quest\022F\n\030rebalance_partition_info\030\001 \002(\0132" + - "$.voldemort.RebalancePartitionInfoMap\"m\n" + - "#InitiateRebalanceNodeOnDonorRequest\022F\n\030", - "rebalance_partition_info\030\001 \003(\0132$.voldemo" + - "rt.RebalancePartitionInfoMap\"\212\001\n\034AsyncOp" + - "erationStatusResponse\022\022\n\nrequest_id\030\001 \001(" + - "\005\022\023\n\013description\030\002 \001(\t\022\016\n\006status\030\003 \001(\t\022\020" + - "\n\010complete\030\004 \001(\010\022\037\n\005error\030\005 \001(\0132\020.voldem" + - "ort.Error\"\'\n\026TruncateEntriesRequest\022\r\n\005s" + - "tore\030\001 \002(\t\":\n\027TruncateEntriesResponse\022\037\n" + - "\005error\030\001 \001(\0132\020.voldemort.Error\"*\n\017AddSto" + - "reRequest\022\027\n\017storeDefinition\030\001 \002(\t\"3\n\020Ad" + - "dStoreResponse\022\037\n\005error\030\001 \001(\0132\020.voldemor", - "t.Error\"\'\n\022DeleteStoreRequest\022\021\n\tstoreNa" + - "me\030\001 \002(\t\"6\n\023DeleteStoreResponse\022\037\n\005error" + - "\030\001 \001(\0132\020.voldemort.Error\"P\n\021FetchStoreRe" + - "quest\022\022\n\nstore_name\030\001 \002(\t\022\021\n\tstore_dir\030\002" + - " \002(\t\022\024\n\014push_version\030\003 \001(\003\"9\n\020SwapStoreR" + - "equest\022\022\n\nstore_name\030\001 \002(\t\022\021\n\tstore_dir\030" + - "\002 \002(\t\"P\n\021SwapStoreResponse\022\037\n\005error\030\001 \001(" + - "\0132\020.voldemort.Error\022\032\n\022previous_store_di" + - "r\030\002 \001(\t\"@\n\024RollbackStoreRequest\022\022\n\nstore" + - "_name\030\001 \002(\t\022\024\n\014push_version\030\002 \002(\003\"8\n\025Rol", - "lbackStoreResponse\022\037\n\005error\030\001 \001(\0132\020.vold" + - "emort.Error\"&\n\020RepairJobRequest\022\022\n\nstore" + - "_name\030\001 \001(\t\"4\n\021RepairJobResponse\022\037\n\005erro" + - "r\030\001 \001(\0132\020.voldemort.Error\"=\n\024ROStoreVers" + - "ionDirMap\022\022\n\nstore_name\030\001 \002(\t\022\021\n\tstore_d" + - "ir\030\002 \002(\t\"/\n\031GetROMaxVersionDirRequest\022\022\n" + - "\nstore_name\030\001 \003(\t\"y\n\032GetROMaxVersionDirR" + - "esponse\022:\n\021ro_store_versions\030\001 \003(\0132\037.vol" + - "demort.ROStoreVersionDirMap\022\037\n\005error\030\002 \001" + - "(\0132\020.voldemort.Error\"3\n\035GetROCurrentVers", - "ionDirRequest\022\022\n\nstore_name\030\001 \003(\t\"}\n\036Get" + - "ROCurrentVersionDirResponse\022:\n\021ro_store_" + - "versions\030\001 \003(\0132\037.voldemort.ROStoreVersio" + - "nDirMap\022\037\n\005error\030\002 \001(\0132\020.voldemort.Error" + - "\"/\n\031GetROStorageFormatRequest\022\022\n\nstore_n" + - "ame\030\001 \003(\t\"y\n\032GetROStorageFormatResponse\022" + - ":\n\021ro_store_versions\030\001 \003(\0132\037.voldemort.R" + - "OStoreVersionDirMap\022\037\n\005error\030\002 \001(\0132\020.vol" + - "demort.Error\"@\n\027FailedFetchStoreRequest\022" + - "\022\n\nstore_name\030\001 \002(\t\022\021\n\tstore_dir\030\002 \002(\t\";", - "\n\030FailedFetchStoreResponse\022\037\n\005error\030\001 \001(" + - "\0132\020.voldemort.Error\"\346\001\n\033RebalanceStateCh" + - "angeRequest\022K\n\035rebalance_partition_info_" + - "list\030\001 \003(\0132$.voldemort.RebalancePartitio" + - "nInfoMap\022\026\n\016cluster_string\030\002 \002(\t\022\017\n\007swap" + - "_ro\030\003 \002(\010\022\037\n\027change_cluster_metadata\030\004 \002" + - "(\010\022\036\n\026change_rebalance_state\030\005 \002(\010\022\020\n\010ro" + - "llback\030\006 \002(\010\"?\n\034RebalanceStateChangeResp" + - "onse\022\037\n\005error\030\001 \001(\0132\020.voldemort.Error\"G\n" + - " DeleteStoreRebalanceStateRequest\022\022\n\nsto", - "re_name\030\001 \002(\t\022\017\n\007node_id\030\002 \002(\005\"D\n!Delete" + - "StoreRebalanceStateResponse\022\037\n\005error\030\001 \001" + - "(\0132\020.voldemort.Error\"h\n\023NativeBackupRequ" + - "est\022\022\n\nstore_name\030\001 \002(\t\022\022\n\nbackup_dir\030\002 " + - "\002(\t\022\024\n\014verify_files\030\003 \002(\010\022\023\n\013incremental" + - "\030\004 \002(\010\">\n\024ReserveMemoryRequest\022\022\n\nstore_" + - "name\030\001 \002(\t\022\022\n\nsize_in_mb\030\002 \002(\003\"8\n\025Reserv" + - "eMemoryResponse\022\037\n\005error\030\001 \001(\0132\020.voldemo" + - "rt.Error\"\360\016\n\025VoldemortAdminRequest\022)\n\004ty" + - "pe\030\001 \002(\0162\033.voldemort.AdminRequestType\0223\n", - "\014get_metadata\030\002 \001(\0132\035.voldemort.GetMetad" + - "ataRequest\0229\n\017update_metadata\030\003 \001(\0132 .vo" + - "ldemort.UpdateMetadataRequest\022J\n\030update_" + - "partition_entries\030\004 \001(\0132(.voldemort.Upda" + - "tePartitionEntriesRequest\022H\n\027fetch_parti" + - "tion_entries\030\005 \001(\0132\'.voldemort.FetchPart" + - "itionEntriesRequest\022J\n\030delete_partition_" + - "entries\030\006 \001(\0132(.voldemort.DeletePartitio" + - "nEntriesRequest\022K\n\031initiate_fetch_and_up" + - "date\030\007 \001(\0132(.voldemort.InitiateFetchAndU", - "pdateRequest\022F\n\026async_operation_status\030\010" + - " \001(\0132&.voldemort.AsyncOperationStatusReq" + - "uest\022H\n\027initiate_rebalance_node\030\t \001(\0132\'." + - "voldemort.InitiateRebalanceNodeRequest\022B" + - "\n\024async_operation_stop\030\n \001(\0132$.voldemort" + - ".AsyncOperationStopRequest\022B\n\024async_oper" + - "ation_list\030\013 \001(\0132$.voldemort.AsyncOperat" + - "ionListRequest\022;\n\020truncate_entries\030\014 \001(\013" + - "2!.voldemort.TruncateEntriesRequest\022-\n\ta" + - "dd_store\030\r \001(\0132\032.voldemort.AddStoreReque", - "st\0223\n\014delete_store\030\016 \001(\0132\035.voldemort.Del" + - "eteStoreRequest\0221\n\013fetch_store\030\017 \001(\0132\034.v" + - "oldemort.FetchStoreRequest\022/\n\nswap_store" + - "\030\020 \001(\0132\033.voldemort.SwapStoreRequest\0227\n\016r" + - "ollback_store\030\021 \001(\0132\037.voldemort.Rollback" + - "StoreRequest\022D\n\026get_ro_max_version_dir\030\022" + - " \001(\0132$.voldemort.GetROMaxVersionDirReque" + - "st\022L\n\032get_ro_current_version_dir\030\023 \001(\0132(" + - ".voldemort.GetROCurrentVersionDirRequest" + - "\022D\n\025fetch_partition_files\030\024 \001(\0132%.voldem", - "ort.FetchPartitionFilesRequest\022@\n\023update" + - "_slop_entries\030\026 \001(\0132#.voldemort.UpdateSl" + - "opEntriesRequest\022>\n\022failed_fetch_store\030\030" + - " \001(\0132\".voldemort.FailedFetchStoreRequest" + - "\022C\n\025get_ro_storage_format\030\031 \001(\0132$.voldem" + - "ort.GetROStorageFormatRequest\022F\n\026rebalan" + - "ce_state_change\030\032 \001(\0132&.voldemort.Rebala" + - "nceStateChangeRequest\022/\n\nrepair_job\030\033 \001(" + - "\0132\033.voldemort.RepairJobRequest\022X\n initia" + - "te_rebalance_node_on_donor\030\034 \001(\0132..volde", - "mort.InitiateRebalanceNodeOnDonorRequest" + - "\022Q\n\034delete_store_rebalance_state\030\035 \001(\0132+" + - ".voldemort.DeleteStoreRebalanceStateRequ" + - "est\0225\n\rnative_backup\030\036 \001(\0132\036.voldemort.N" + - "ativeBackupRequest\0227\n\016reserve_memory\030\037 \001" + - "(\0132\037.voldemort.ReserveMemoryRequest*\310\005\n\020" + - "AdminRequestType\022\020\n\014GET_METADATA\020\000\022\023\n\017UP" + - "DATE_METADATA\020\001\022\034\n\030UPDATE_PARTITION_ENTR" + - "IES\020\002\022\033\n\027FETCH_PARTITION_ENTRIES\020\003\022\034\n\030DE" + - "LETE_PARTITION_ENTRIES\020\004\022\035\n\031INITIATE_FET", - "CH_AND_UPDATE\020\005\022\032\n\026ASYNC_OPERATION_STATU" + - "S\020\006\022\033\n\027INITIATE_REBALANCE_NODE\020\007\022\030\n\024ASYN" + - "C_OPERATION_STOP\020\010\022\030\n\024ASYNC_OPERATION_LI" + - "ST\020\t\022\024\n\020TRUNCATE_ENTRIES\020\n\022\r\n\tADD_STORE\020" + - "\013\022\020\n\014DELETE_STORE\020\014\022\017\n\013FETCH_STORE\020\r\022\016\n\n" + - "SWAP_STORE\020\016\022\022\n\016ROLLBACK_STORE\020\017\022\032\n\026GET_" + - "RO_MAX_VERSION_DIR\020\020\022\036\n\032GET_RO_CURRENT_V" + - "ERSION_DIR\020\021\022\031\n\025FETCH_PARTITION_FILES\020\022\022" + - "\027\n\023UPDATE_SLOP_ENTRIES\020\024\022\026\n\022FAILED_FETCH" + - "_STORE\020\026\022\031\n\025GET_RO_STORAGE_FORMAT\020\027\022\032\n\026R", - "EBALANCE_STATE_CHANGE\020\030\022\016\n\nREPAIR_JOB\020\031\022" + - "$\n INITIATE_REBALANCE_NODE_ON_DONOR\020\032\022 \n" + - "\034DELETE_STORE_REBALANCE_STATE\020\033\022\021\n\rNATIV" + - "E_BACKUP\020\034\022\022\n\016RESERVE_MEMORY\020\035B-\n\034voldem" + - "ort.client.protocol.pbB\013VAdminProtoH\001" + "\022\035\n\025records_per_partition\030\010 \001(\003\"\201\001\n\035Fetc" + + "hPartitionEntriesResponse\0222\n\017partition_e" + + "ntry\030\001 \001(\0132\031.voldemort.PartitionEntry\022\013\n" + + "\003key\030\002 \001(\014\022\037\n\005error\030\003 \001(\0132\020.voldemort.Er" + + "ror\"\254\001\n\035DeletePartitionEntriesRequest\022\r\n" + + "\005store\030\001 \002(\t\0227\n\024replica_to_partition\030\002 \003" + + "(\0132\031.voldemort.PartitionTuple\022*\n\006filter\030", + "\003 \001(\0132\032.voldemort.VoldemortFilter\022\027\n\017ini" + + "tial_cluster\030\004 \001(\t\"P\n\036DeletePartitionEnt" + + "riesResponse\022\r\n\005count\030\001 \001(\003\022\037\n\005error\030\002 \001" + + "(\0132\020.voldemort.Error\"\317\001\n\035InitiateFetchAn" + + "dUpdateRequest\022\017\n\007node_id\030\001 \002(\005\022\r\n\005store" + + "\030\002 \002(\t\022*\n\006filter\030\003 \001(\0132\032.voldemort.Volde" + + "mortFilter\0227\n\024replica_to_partition\030\004 \003(\013" + + "2\031.voldemort.PartitionTuple\022\027\n\017initial_c" + + "luster\030\005 \001(\t\022\020\n\010optimize\030\006 \001(\010\"1\n\033AsyncO" + + "perationStatusRequest\022\022\n\nrequest_id\030\001 \002(", + "\005\"/\n\031AsyncOperationStopRequest\022\022\n\nreques" + + "t_id\030\001 \002(\005\"=\n\032AsyncOperationStopResponse" + + "\022\037\n\005error\030\001 \001(\0132\020.voldemort.Error\"2\n\031Asy" + + "ncOperationListRequest\022\025\n\rshow_complete\030" + + "\002 \002(\010\"R\n\032AsyncOperationListResponse\022\023\n\013r" + + "equest_ids\030\001 \003(\005\022\037\n\005error\030\002 \001(\0132\020.voldem" + + "ort.Error\":\n\016PartitionTuple\022\024\n\014replica_t" + + "ype\030\001 \002(\005\022\022\n\npartitions\030\002 \003(\005\"e\n\026PerStor" + + "ePartitionTuple\022\022\n\nstore_name\030\001 \002(\t\0227\n\024r" + + "eplica_to_partition\030\002 \003(\0132\031.voldemort.Pa", + "rtitionTuple\"\370\001\n\031RebalancePartitionInfoM" + + "ap\022\022\n\nstealer_id\030\001 \002(\005\022\020\n\010donor_id\030\002 \002(\005" + + "\022\017\n\007attempt\030\003 \002(\005\022C\n\030replica_to_add_part" + + "ition\030\004 \003(\0132!.voldemort.PerStorePartitio" + + "nTuple\022F\n\033replica_to_delete_partition\030\005 " + + "\003(\0132!.voldemort.PerStorePartitionTuple\022\027" + + "\n\017initial_cluster\030\006 \002(\t\"f\n\034InitiateRebal" + + "anceNodeRequest\022F\n\030rebalance_partition_i" + + "nfo\030\001 \002(\0132$.voldemort.RebalancePartition" + + "InfoMap\"m\n#InitiateRebalanceNodeOnDonorR", + "equest\022F\n\030rebalance_partition_info\030\001 \003(\013" + + "2$.voldemort.RebalancePartitionInfoMap\"\212" + + "\001\n\034AsyncOperationStatusResponse\022\022\n\nreque" + + "st_id\030\001 \001(\005\022\023\n\013description\030\002 \001(\t\022\016\n\006stat" + + "us\030\003 \001(\t\022\020\n\010complete\030\004 \001(\010\022\037\n\005error\030\005 \001(" + + "\0132\020.voldemort.Error\"\'\n\026TruncateEntriesRe" + + "quest\022\r\n\005store\030\001 \002(\t\":\n\027TruncateEntriesR" + + "esponse\022\037\n\005error\030\001 \001(\0132\020.voldemort.Error" + + "\"*\n\017AddStoreRequest\022\027\n\017storeDefinition\030\001" + + " \002(\t\"3\n\020AddStoreResponse\022\037\n\005error\030\001 \001(\0132", + "\020.voldemort.Error\"\'\n\022DeleteStoreRequest\022" + + "\021\n\tstoreName\030\001 \002(\t\"6\n\023DeleteStoreRespons" + + "e\022\037\n\005error\030\001 \001(\0132\020.voldemort.Error\"P\n\021Fe" + + "tchStoreRequest\022\022\n\nstore_name\030\001 \002(\t\022\021\n\ts" + + "tore_dir\030\002 \002(\t\022\024\n\014push_version\030\003 \001(\003\"9\n\020" + + "SwapStoreRequest\022\022\n\nstore_name\030\001 \002(\t\022\021\n\t" + + "store_dir\030\002 \002(\t\"P\n\021SwapStoreResponse\022\037\n\005" + + "error\030\001 \001(\0132\020.voldemort.Error\022\032\n\022previou" + + "s_store_dir\030\002 \001(\t\"@\n\024RollbackStoreReques" + + "t\022\022\n\nstore_name\030\001 \002(\t\022\024\n\014push_version\030\002 ", + "\002(\003\"8\n\025RollbackStoreResponse\022\037\n\005error\030\001 " + + "\001(\0132\020.voldemort.Error\"&\n\020RepairJobReques" + + "t\022\022\n\nstore_name\030\001 \001(\t\"4\n\021RepairJobRespon" + + "se\022\037\n\005error\030\001 \001(\0132\020.voldemort.Error\"=\n\024R" + + "OStoreVersionDirMap\022\022\n\nstore_name\030\001 \002(\t\022" + + "\021\n\tstore_dir\030\002 \002(\t\"/\n\031GetROMaxVersionDir" + + "Request\022\022\n\nstore_name\030\001 \003(\t\"y\n\032GetROMaxV" + + "ersionDirResponse\022:\n\021ro_store_versions\030\001" + + " \003(\0132\037.voldemort.ROStoreVersionDirMap\022\037\n" + + "\005error\030\002 \001(\0132\020.voldemort.Error\"3\n\035GetROC", + "urrentVersionDirRequest\022\022\n\nstore_name\030\001 " + + "\003(\t\"}\n\036GetROCurrentVersionDirResponse\022:\n" + + "\021ro_store_versions\030\001 \003(\0132\037.voldemort.ROS" + + "toreVersionDirMap\022\037\n\005error\030\002 \001(\0132\020.volde" + + "mort.Error\"/\n\031GetROStorageFormatRequest\022" + + "\022\n\nstore_name\030\001 \003(\t\"y\n\032GetROStorageForma" + + "tResponse\022:\n\021ro_store_versions\030\001 \003(\0132\037.v" + + "oldemort.ROStoreVersionDirMap\022\037\n\005error\030\002" + + " \001(\0132\020.voldemort.Error\"@\n\027FailedFetchSto" + + "reRequest\022\022\n\nstore_name\030\001 \002(\t\022\021\n\tstore_d", + "ir\030\002 \002(\t\";\n\030FailedFetchStoreResponse\022\037\n\005" + + "error\030\001 \001(\0132\020.voldemort.Error\"\346\001\n\033Rebala" + + "nceStateChangeRequest\022K\n\035rebalance_parti" + + "tion_info_list\030\001 \003(\0132$.voldemort.Rebalan" + + "cePartitionInfoMap\022\026\n\016cluster_string\030\002 \002" + + "(\t\022\017\n\007swap_ro\030\003 \002(\010\022\037\n\027change_cluster_me" + + "tadata\030\004 \002(\010\022\036\n\026change_rebalance_state\030\005" + + " \002(\010\022\020\n\010rollback\030\006 \002(\010\"?\n\034RebalanceState" + + "ChangeResponse\022\037\n\005error\030\001 \001(\0132\020.voldemor" + + "t.Error\"G\n DeleteStoreRebalanceStateRequ", + "est\022\022\n\nstore_name\030\001 \002(\t\022\017\n\007node_id\030\002 \002(\005" + + "\"D\n!DeleteStoreRebalanceStateResponse\022\037\n" + + "\005error\030\001 \001(\0132\020.voldemort.Error\"h\n\023Native" + + "BackupRequest\022\022\n\nstore_name\030\001 \002(\t\022\022\n\nbac" + + "kup_dir\030\002 \002(\t\022\024\n\014verify_files\030\003 \002(\010\022\023\n\013i" + + "ncremental\030\004 \002(\010\">\n\024ReserveMemoryRequest" + + "\022\022\n\nstore_name\030\001 \002(\t\022\022\n\nsize_in_mb\030\002 \002(\003" + + "\"8\n\025ReserveMemoryResponse\022\037\n\005error\030\001 \001(\013" + + "2\020.voldemort.Error\"\360\016\n\025VoldemortAdminReq" + + "uest\022)\n\004type\030\001 \002(\0162\033.voldemort.AdminRequ", + "estType\0223\n\014get_metadata\030\002 \001(\0132\035.voldemor" + + "t.GetMetadataRequest\0229\n\017update_metadata\030" + + "\003 \001(\0132 .voldemort.UpdateMetadataRequest\022" + + "J\n\030update_partition_entries\030\004 \001(\0132(.vold" + + "emort.UpdatePartitionEntriesRequest\022H\n\027f" + + "etch_partition_entries\030\005 \001(\0132\'.voldemort" + + ".FetchPartitionEntriesRequest\022J\n\030delete_" + + "partition_entries\030\006 \001(\0132(.voldemort.Dele" + + "tePartitionEntriesRequest\022K\n\031initiate_fe" + + "tch_and_update\030\007 \001(\0132(.voldemort.Initiat", + "eFetchAndUpdateRequest\022F\n\026async_operatio" + + "n_status\030\010 \001(\0132&.voldemort.AsyncOperatio" + + "nStatusRequest\022H\n\027initiate_rebalance_nod" + + "e\030\t \001(\0132\'.voldemort.InitiateRebalanceNod" + + "eRequest\022B\n\024async_operation_stop\030\n \001(\0132$" + + ".voldemort.AsyncOperationStopRequest\022B\n\024" + + "async_operation_list\030\013 \001(\0132$.voldemort.A" + + "syncOperationListRequest\022;\n\020truncate_ent" + + "ries\030\014 \001(\0132!.voldemort.TruncateEntriesRe" + + "quest\022-\n\tadd_store\030\r \001(\0132\032.voldemort.Add", + "StoreRequest\0223\n\014delete_store\030\016 \001(\0132\035.vol" + + "demort.DeleteStoreRequest\0221\n\013fetch_store" + + "\030\017 \001(\0132\034.voldemort.FetchStoreRequest\022/\n\n" + + "swap_store\030\020 \001(\0132\033.voldemort.SwapStoreRe" + + "quest\0227\n\016rollback_store\030\021 \001(\0132\037.voldemor" + + "t.RollbackStoreRequest\022D\n\026get_ro_max_ver" + + "sion_dir\030\022 \001(\0132$.voldemort.GetROMaxVersi" + + "onDirRequest\022L\n\032get_ro_current_version_d" + + "ir\030\023 \001(\0132(.voldemort.GetROCurrentVersion" + + "DirRequest\022D\n\025fetch_partition_files\030\024 \001(", + "\0132%.voldemort.FetchPartitionFilesRequest" + + "\022@\n\023update_slop_entries\030\026 \001(\0132#.voldemor" + + "t.UpdateSlopEntriesRequest\022>\n\022failed_fet" + + "ch_store\030\030 \001(\0132\".voldemort.FailedFetchSt" + + "oreRequest\022C\n\025get_ro_storage_format\030\031 \001(" + + "\0132$.voldemort.GetROStorageFormatRequest\022" + + "F\n\026rebalance_state_change\030\032 \001(\0132&.voldem" + + "ort.RebalanceStateChangeRequest\022/\n\nrepai" + + "r_job\030\033 \001(\0132\033.voldemort.RepairJobRequest" + + "\022X\n initiate_rebalance_node_on_donor\030\034 \001", + "(\0132..voldemort.InitiateRebalanceNodeOnDo" + + "norRequest\022Q\n\034delete_store_rebalance_sta" + + "te\030\035 \001(\0132+.voldemort.DeleteStoreRebalanc" + + "eStateRequest\0225\n\rnative_backup\030\036 \001(\0132\036.v" + + "oldemort.NativeBackupRequest\0227\n\016reserve_" + + "memory\030\037 \001(\0132\037.voldemort.ReserveMemoryRe" + + "quest*\310\005\n\020AdminRequestType\022\020\n\014GET_METADA" + + "TA\020\000\022\023\n\017UPDATE_METADATA\020\001\022\034\n\030UPDATE_PART" + + "ITION_ENTRIES\020\002\022\033\n\027FETCH_PARTITION_ENTRI" + + "ES\020\003\022\034\n\030DELETE_PARTITION_ENTRIES\020\004\022\035\n\031IN", + "ITIATE_FETCH_AND_UPDATE\020\005\022\032\n\026ASYNC_OPERA" + + "TION_STATUS\020\006\022\033\n\027INITIATE_REBALANCE_NODE" + + "\020\007\022\030\n\024ASYNC_OPERATION_STOP\020\010\022\030\n\024ASYNC_OP" + + "ERATION_LIST\020\t\022\024\n\020TRUNCATE_ENTRIES\020\n\022\r\n\t" + + "ADD_STORE\020\013\022\020\n\014DELETE_STORE\020\014\022\017\n\013FETCH_S" + + "TORE\020\r\022\016\n\nSWAP_STORE\020\016\022\022\n\016ROLLBACK_STORE" + + "\020\017\022\032\n\026GET_RO_MAX_VERSION_DIR\020\020\022\036\n\032GET_RO" + + "_CURRENT_VERSION_DIR\020\021\022\031\n\025FETCH_PARTITIO" + + "N_FILES\020\022\022\027\n\023UPDATE_SLOP_ENTRIES\020\024\022\026\n\022FA" + + "ILED_FETCH_STORE\020\026\022\031\n\025GET_RO_STORAGE_FOR", + "MAT\020\027\022\032\n\026REBALANCE_STATE_CHANGE\020\030\022\016\n\nREP" + + "AIR_JOB\020\031\022$\n INITIATE_REBALANCE_NODE_ON_" + + "DONOR\020\032\022 \n\034DELETE_STORE_REBALANCE_STATE\020" + + "\033\022\021\n\rNATIVE_BACKUP\020\034\022\022\n\016RESERVE_MEMORY\020\035" + + "B-\n\034voldemort.client.protocol.pbB\013VAdmin" + + "ProtoH\001" }; com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner = new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() { @@ -23439,7 +23440,7 @@ public com.google.protobuf.ExtensionRegistry assignDescriptors( internal_static_voldemort_FetchPartitionEntriesRequest_fieldAccessorTable = new com.google.protobuf.GeneratedMessage.FieldAccessorTable( internal_static_voldemort_FetchPartitionEntriesRequest_descriptor, - new java.lang.String[] { "ReplicaToPartition", "Store", "Filter", "FetchValues", "OBSOLETEDONOTUSESkipRecords", "InitialCluster", "FetchOrphaned", "MaxRecords", }, + new java.lang.String[] { "ReplicaToPartition", "Store", "Filter", "FetchValues", "OBSOLETEDONOTUSESkipRecords", "InitialCluster", "FetchOrphaned", "RecordsPerPartition", }, voldemort.client.protocol.pb.VAdminProto.FetchPartitionEntriesRequest.class, voldemort.client.protocol.pb.VAdminProto.FetchPartitionEntriesRequest.Builder.class); internal_static_voldemort_FetchPartitionEntriesResponse_descriptor = diff --git a/src/java/voldemort/server/protocol/admin/FetchEntriesStreamRequestHandler.java b/src/java/voldemort/server/protocol/admin/FetchEntriesStreamRequestHandler.java index f8637c9041..4a418d9b42 100644 --- a/src/java/voldemort/server/protocol/admin/FetchEntriesStreamRequestHandler.java +++ b/src/java/voldemort/server/protocol/admin/FetchEntriesStreamRequestHandler.java @@ -68,6 +68,7 @@ public FetchEntriesStreamRequestHandler(FetchPartitionEntriesRequest request, + "' with replica to partition mapping " + replicaToPartitionList); } + @Override public StreamRequestHandlerState handleRequest(DataInputStream inputStream, DataOutputStream outputStream) throws IOException { @@ -142,7 +143,10 @@ public StreamRequestHandlerState handleRequest(DataInputStream inputStream, + " s"); } - if(keyIterator.hasNext() && counter < maxRecords) + // TODO: Add logic to FetchKeys and FetchEntries to count keys per + // partition correctly. + + if(keyIterator.hasNext() && counter < recordsPerPartition) return StreamRequestHandlerState.WRITING; else { return StreamRequestHandlerState.COMPLETE; diff --git a/src/java/voldemort/server/protocol/admin/FetchKeysStreamRequestHandler.java b/src/java/voldemort/server/protocol/admin/FetchKeysStreamRequestHandler.java index 460f7e6190..407984b0d2 100644 --- a/src/java/voldemort/server/protocol/admin/FetchKeysStreamRequestHandler.java +++ b/src/java/voldemort/server/protocol/admin/FetchKeysStreamRequestHandler.java @@ -127,9 +127,9 @@ public StreamRequestHandlerState handleRequest(DataInputStream inputStream, // TODO: Make sure the distinction between FetchKeysStream and // FetchPartitionKeysStream is clear. - // TODO: Add logic to FetchKeys and FetchEntries to account for keys per - // partition. - if(keyIterator.hasNext() && (counter < maxRecords)) + // TODO: Add logic to FetchKeys and FetchEntries to count keys per + // partition correctly. + if(keyIterator.hasNext() && (counter < recordsPerPartition)) return StreamRequestHandlerState.WRITING; else { return StreamRequestHandlerState.COMPLETE; diff --git a/src/java/voldemort/server/protocol/admin/FetchPartitionEntriesStreamRequestHandler.java b/src/java/voldemort/server/protocol/admin/FetchPartitionEntriesStreamRequestHandler.java index 58fde9efc3..d3419e3dde 100644 --- a/src/java/voldemort/server/protocol/admin/FetchPartitionEntriesStreamRequestHandler.java +++ b/src/java/voldemort/server/protocol/admin/FetchPartitionEntriesStreamRequestHandler.java @@ -86,13 +86,14 @@ public FetchPartitionEntriesStreamRequestHandler(FetchPartitionEntriesRequest re } } + @Override public StreamRequestHandlerState handleRequest(DataInputStream inputStream, DataOutputStream outputStream) throws IOException { // process the next partition if(entriesPartitionIterator == null) { - if(currentIndex == partitionList.size() || counter >= maxRecords) { + if(currentIndex == partitionList.size() || counter >= recordsPerPartition) { // TODO: Make .info consistent logger.info("Done fetching store " + storageEngine.getName() + " : " + counter + " records processed."); @@ -168,8 +169,11 @@ public StreamRequestHandlerState handleRequest(DataInputStream inputStream, } } + // TODO: Add logic to FetchKeys and FetchEntries to count keys per + // partition correctly. + // reset the iterator if done with this partition - if(!entriesPartitionIterator.hasNext() || counter >= maxRecords) { + if(!entriesPartitionIterator.hasNext() || counter >= recordsPerPartition) { entriesPartitionIterator.close(); entriesPartitionIterator = null; } diff --git a/src/java/voldemort/server/protocol/admin/FetchPartitionKeysStreamRequestHandler.java b/src/java/voldemort/server/protocol/admin/FetchPartitionKeysStreamRequestHandler.java index a3f2e2cbbe..55b51b29a6 100644 --- a/src/java/voldemort/server/protocol/admin/FetchPartitionKeysStreamRequestHandler.java +++ b/src/java/voldemort/server/protocol/admin/FetchPartitionKeysStreamRequestHandler.java @@ -91,7 +91,7 @@ public StreamRequestHandlerState handleRequest(DataInputStream inputStream, // process the next partition if(keysPartitionIterator == null) { - if(currentIndex == partitionList.size() || counter >= maxRecords) { + if(currentIndex == partitionList.size() || counter >= recordsPerPartition) { // TODO: Make all .info messages consistent. "Records fetched" // instead of "Done fetching". logger.info("Done fetching store " + storageEngine.getName() + " : " + counter @@ -164,9 +164,12 @@ public StreamRequestHandlerState handleRequest(DataInputStream inputStream, } } + // TODO: Add logic to FetchKeys and FetchEntries to count keys per + // partition correctly. + // reset the iterator if done with this partition or fetched enough // records - if(!keysPartitionIterator.hasNext() || (counter >= maxRecords)) { + if(!keysPartitionIterator.hasNext() || (counter >= recordsPerPartition)) { keysPartitionIterator.close(); keysPartitionIterator = null; } diff --git a/src/java/voldemort/server/protocol/admin/FetchStreamRequestHandler.java b/src/java/voldemort/server/protocol/admin/FetchStreamRequestHandler.java index 196e105675..727289e0fd 100644 --- a/src/java/voldemort/server/protocol/admin/FetchStreamRequestHandler.java +++ b/src/java/voldemort/server/protocol/admin/FetchStreamRequestHandler.java @@ -68,7 +68,7 @@ public abstract class FetchStreamRequestHandler implements StreamRequestHandler protected long counter; - protected long maxRecords; + protected long recordsPerPartition; protected int fetched; @@ -120,11 +120,9 @@ protected FetchStreamRequestHandler(VAdminProto.FetchPartitionEntriesRequest req this.startTime = System.currentTimeMillis(); this.counter = 0; - // TODO: maxRecords should default to 0 for clarity imho. (And change to - // recordsPerPartition_ - this.maxRecords = Long.MAX_VALUE; - if(request.hasMaxRecords() && request.getMaxRecords() > 0) { - this.maxRecords = request.getMaxRecords(); + this.recordsPerPartition = 0; + if(request.hasRecordsPerPartition() && request.getRecordsPerPartition() > 0) { + this.recordsPerPartition = request.getRecordsPerPartition(); } this.fetchOrphaned = request.hasFetchOrphaned() && request.getFetchOrphaned(); } @@ -139,10 +137,12 @@ private StoreDefinition getStoreDef(String store, MetadataStore metadataStore) { return def; } + @Override public final StreamRequestDirection getDirection() { return StreamRequestDirection.WRITING; } + @Override public void close(DataOutputStream outputStream) throws IOException { logger.info("Successfully scanned " + counter + " tuples, fetched " + fetched + " tuples for store '" + storageEngine.getName() + "' in " @@ -151,6 +151,7 @@ public void close(DataOutputStream outputStream) throws IOException { ProtoUtils.writeEndOfStream(outputStream); } + @Override public final void handleError(DataOutputStream outputStream, VoldemortException e) throws IOException { VAdminProto.FetchPartitionEntriesResponse response = VAdminProto.FetchPartitionEntriesResponse.newBuilder() diff --git a/src/java/voldemort/utils/Entropy.java b/src/java/voldemort/utils/Entropy.java index 7c23194d6c..8d38b3fa48 100644 --- a/src/java/voldemort/utils/Entropy.java +++ b/src/java/voldemort/utils/Entropy.java @@ -48,6 +48,9 @@ import com.google.common.base.Joiner; import com.google.common.collect.Maps; +// Entropy is replaced by KeySamplerCLI and KeyVersionFetcherCLI. Entropy never +// really worked as described and had a complicated interface. +@Deprecated public class Entropy { private int nodeId; diff --git a/src/java/voldemort/utils/KeySamplerCLI.java b/src/java/voldemort/utils/KeySamplerCLI.java index 748eb83e75..e63d6a5caa 100644 --- a/src/java/voldemort/utils/KeySamplerCLI.java +++ b/src/java/voldemort/utils/KeySamplerCLI.java @@ -56,7 +56,7 @@ public class KeySamplerCLI { private static Logger logger = Logger.getLogger(KeySamplerCLI.class); private final static int DEFAULT_NODE_PARALLELISM = 8; - private final static int DEFAULT_MAX_RECORDS = 1; + private final static int DEFAULT_RECORDS_PER_PARTITION = 1; private final AdminClient adminClient; private final Cluster cluster; @@ -66,9 +66,9 @@ public class KeySamplerCLI { private final String outDir; private final ExecutorService nodeSamplerService; - private final int maxRecords; + private final int recordsPerPartition; - public KeySamplerCLI(String url, String outDir, int nodeParallelism, int maxRecords) { + public KeySamplerCLI(String url, String outDir, int nodeParallelism, int recordsPerPartition) { if(logger.isInfoEnabled()) { logger.info("Connecting to bootstrap server: " + url); } @@ -84,7 +84,7 @@ public KeySamplerCLI(String url, String outDir, int nodeParallelism, int maxReco this.nodeSamplerService = Executors.newFixedThreadPool(nodeParallelism); - this.maxRecords = maxRecords; + this.recordsPerPartition = recordsPerPartition; } public boolean sampleStores() { @@ -126,11 +126,14 @@ public NodeSampleResult call() throws Exception { String storeName = storeDefinition.getName(); StringBuilder hexKeyStrings = new StringBuilder(); + // TODO: Change this from a loop to flat st the list of partitoinIds + // are sent to server. for(int partitionId: node.getPartitionIds()) { success = false; // TODO: real per-server throttling and/or make '100' a command - // line argument. + // line argument. And, move near .next() to throttle server + // suckage. // Simple, lame throttling since thread is going at same node // repeatedly @@ -159,7 +162,7 @@ public NodeSampleResult call() throws Exception { singlePartition, null, true, - maxRecords); + recordsPerPartition); int keyCount = 0; while(fetchIterator.hasNext()) { ByteArray key = fetchIterator.next(); @@ -167,12 +170,12 @@ public NodeSampleResult call() throws Exception { hexKeyStrings.append(hexKeyString + "\n"); keyCount++; } - if(keyCount < maxRecords) { + if(keyCount < recordsPerPartition) { logger.warn("Fewer keys (" + keyCount + ") than requested (" - + maxRecords + ") returned --- " + infoTag); - } else if(keyCount < maxRecords) { + + recordsPerPartition + ") returned --- " + infoTag); + } else if(keyCount < recordsPerPartition) { logger.warn("More keys (" + keyCount + ") than requested (" - + maxRecords + ") returned --- " + infoTag); + + recordsPerPartition + ") returned --- " + infoTag); } success = true; } catch(VoldemortException ve) { @@ -291,11 +294,11 @@ private static OptionParser getParser() { .withRequiredArg() .describedAs("storeParallelism") .ofType(Integer.class); - parser.accepts("max-records", - "Number of keys sampled per partitoin. [Default: " + DEFAULT_MAX_RECORDS - + " ]") + parser.accepts("records-per-partition", + "Number of keys sampled per partition. [Default: " + + DEFAULT_RECORDS_PER_PARTITION + " ]") .withRequiredArg() - .describedAs("maxRecords") + .describedAs("recordsPerPartition") .ofType(Integer.class); return parser; } @@ -313,7 +316,7 @@ private static void printUsage() { help.append(" --out-dir \n"); help.append(" Optional:\n"); help.append(" --parallelism \n"); - help.append(" --max-records \n"); + help.append(" --records-per-partition \n"); help.append(" --help\n"); System.out.print(help.toString()); } @@ -358,9 +361,9 @@ public static void main(String[] args) throws Exception { nodeParallelism = (Integer) options.valueOf("parallelism"); } - Integer maxRecords = DEFAULT_MAX_RECORDS; - if(options.hasArgument("max-records")) { - maxRecords = (Integer) options.valueOf("max-records"); + Integer recordsPerPartition = DEFAULT_RECORDS_PER_PARTITION; + if(options.hasArgument("records-per-partition")) { + recordsPerPartition = (Integer) options.valueOf("records-per-partition"); } // TODO: Assuming "right thing" happens server-side, then do not need @@ -370,7 +373,10 @@ public static void main(String[] args) throws Exception { + "Use fo this tool against other types of servers is undefined."); try { - KeySamplerCLI sampler = new KeySamplerCLI(url, outDir, nodeParallelism, maxRecords); + KeySamplerCLI sampler = new KeySamplerCLI(url, + outDir, + nodeParallelism, + recordsPerPartition); try { if(!sampler.sampleStores()) { logger.error("Some stores were not successfully sampled."); diff --git a/src/proto/voldemort-admin.proto b/src/proto/voldemort-admin.proto index 4acdd5a65e..4d8710ff0a 100644 --- a/src/proto/voldemort-admin.proto +++ b/src/proto/voldemort-admin.proto @@ -79,8 +79,7 @@ message FetchPartitionEntriesRequest { optional int64 OBSOLETE__DO_NOT_USE__skip_records = 5; optional string initial_cluster = 6; optional bool fetch_orphaned = 7; - optional int64 max_records = 8; - // optional int64 records_per_partition = 8; + optional int64 records_per_partition = 8; } message FetchPartitionEntriesResponse {