http://git-wip-us.apache.org/repos/asf/hbase-site/blob/dbe3a233/devapidocs/src-html/org/apache/hadoop/hbase/master/RegionServerTracker.html ---------------------------------------------------------------------- diff --git a/devapidocs/src-html/org/apache/hadoop/hbase/master/RegionServerTracker.html b/devapidocs/src-html/org/apache/hadoop/hbase/master/RegionServerTracker.html index 06b90f3..256ed63 100644 --- a/devapidocs/src-html/org/apache/hadoop/hbase/master/RegionServerTracker.html +++ b/devapidocs/src-html/org/apache/hadoop/hbase/master/RegionServerTracker.html @@ -135,64 +135,63 @@ <span class="sourceLineNo">127</span> ServerName serverName = pair.getFirst();<a name="line.127"></a> <span class="sourceLineNo">128</span> RegionServerInfo info = pair.getSecond();<a name="line.128"></a> <span class="sourceLineNo">129</span> regionServers.add(serverName);<a name="line.129"></a> -<span class="sourceLineNo">130</span> ServerMetrics serverMetrics = info != null<a name="line.130"></a> -<span class="sourceLineNo">131</span> ? ServerMetricsBuilder.of(serverName,<a name="line.131"></a> -<span class="sourceLineNo">132</span> VersionInfoUtil.getVersionNumber(info.getVersionInfo()))<a name="line.132"></a> -<span class="sourceLineNo">133</span> : ServerMetricsBuilder.of(serverName);<a name="line.133"></a> -<span class="sourceLineNo">134</span> serverManager.checkAndRecordNewServer(serverName, serverMetrics);<a name="line.134"></a> -<span class="sourceLineNo">135</span> }<a name="line.135"></a> -<span class="sourceLineNo">136</span> serverManager.findOutDeadServersAndProcess(deadServersFromPE, liveServersFromWALDir);<a name="line.136"></a> -<span class="sourceLineNo">137</span> }<a name="line.137"></a> -<span class="sourceLineNo">138</span> }<a name="line.138"></a> -<span class="sourceLineNo">139</span><a name="line.139"></a> -<span class="sourceLineNo">140</span> public void stop() {<a name="line.140"></a> -<span class="sourceLineNo">141</span> executor.shutdownNow();<a name="line.141"></a> -<span class="sourceLineNo">142</span> }<a name="line.142"></a> -<span class="sourceLineNo">143</span><a name="line.143"></a> -<span class="sourceLineNo">144</span> private synchronized void refresh() {<a name="line.144"></a> -<span class="sourceLineNo">145</span> List<String> names;<a name="line.145"></a> -<span class="sourceLineNo">146</span> try {<a name="line.146"></a> -<span class="sourceLineNo">147</span> names = ZKUtil.listChildrenAndWatchForNewChildren(watcher, watcher.getZNodePaths().rsZNode);<a name="line.147"></a> -<span class="sourceLineNo">148</span> } catch (KeeperException e) {<a name="line.148"></a> -<span class="sourceLineNo">149</span> // here we need to abort as we failed to set watcher on the rs node which means that we can<a name="line.149"></a> -<span class="sourceLineNo">150</span> // not track the node deleted evetnt any more.<a name="line.150"></a> -<span class="sourceLineNo">151</span> server.abort("Unexpected zk exception getting RS nodes", e);<a name="line.151"></a> -<span class="sourceLineNo">152</span> return;<a name="line.152"></a> -<span class="sourceLineNo">153</span> }<a name="line.153"></a> -<span class="sourceLineNo">154</span> Set<ServerName> servers =<a name="line.154"></a> -<span class="sourceLineNo">155</span> names.stream().map(ServerName::parseServerName).collect(Collectors.toSet());<a name="line.155"></a> -<span class="sourceLineNo">156</span> for (Iterator<ServerName> iter = regionServers.iterator(); iter.hasNext();) {<a name="line.156"></a> -<span class="sourceLineNo">157</span> ServerName sn = iter.next();<a name="line.157"></a> -<span class="sourceLineNo">158</span> if (!servers.contains(sn)) {<a name="line.158"></a> -<span class="sourceLineNo">159</span> LOG.info("RegionServer ephemeral node deleted, processing expiration [{}]", sn);<a name="line.159"></a> -<span class="sourceLineNo">160</span> serverManager.expireServer(sn);<a name="line.160"></a> -<span class="sourceLineNo">161</span> iter.remove();<a name="line.161"></a> -<span class="sourceLineNo">162</span> }<a name="line.162"></a> -<span class="sourceLineNo">163</span> }<a name="line.163"></a> -<span class="sourceLineNo">164</span> // here we do not need to parse the region server info as it is useless now, we only need the<a name="line.164"></a> -<span class="sourceLineNo">165</span> // server name.<a name="line.165"></a> -<span class="sourceLineNo">166</span> boolean newServerAdded = false;<a name="line.166"></a> -<span class="sourceLineNo">167</span> for (ServerName sn : servers) {<a name="line.167"></a> -<span class="sourceLineNo">168</span> if (regionServers.add(sn)) {<a name="line.168"></a> -<span class="sourceLineNo">169</span> newServerAdded = true;<a name="line.169"></a> -<span class="sourceLineNo">170</span> LOG.info("RegionServer ephemeral node created, adding [" + sn + "]");<a name="line.170"></a> -<span class="sourceLineNo">171</span> }<a name="line.171"></a> -<span class="sourceLineNo">172</span> }<a name="line.172"></a> -<span class="sourceLineNo">173</span> if (newServerAdded && server.isInitialized()) {<a name="line.173"></a> -<span class="sourceLineNo">174</span> // Only call the check to move servers if a RegionServer was added to the cluster; in this<a name="line.174"></a> -<span class="sourceLineNo">175</span> // case it could be a server with a new version so it makes sense to run the check.<a name="line.175"></a> -<span class="sourceLineNo">176</span> server.checkIfShouldMoveSystemRegionAsync();<a name="line.176"></a> -<span class="sourceLineNo">177</span> }<a name="line.177"></a> -<span class="sourceLineNo">178</span> }<a name="line.178"></a> -<span class="sourceLineNo">179</span><a name="line.179"></a> -<span class="sourceLineNo">180</span> @Override<a name="line.180"></a> -<span class="sourceLineNo">181</span> public void nodeChildrenChanged(String path) {<a name="line.181"></a> -<span class="sourceLineNo">182</span> if (path.equals(watcher.getZNodePaths().rsZNode) && !server.isAborted() &&<a name="line.182"></a> -<span class="sourceLineNo">183</span> !server.isStopped()) {<a name="line.183"></a> -<span class="sourceLineNo">184</span> executor.execute(this::refresh);<a name="line.184"></a> -<span class="sourceLineNo">185</span> }<a name="line.185"></a> -<span class="sourceLineNo">186</span> }<a name="line.186"></a> -<span class="sourceLineNo">187</span>}<a name="line.187"></a> +<span class="sourceLineNo">130</span> ServerMetrics serverMetrics = info != null ? ServerMetricsBuilder.of(serverName,<a name="line.130"></a> +<span class="sourceLineNo">131</span> VersionInfoUtil.getVersionNumber(info.getVersionInfo()),<a name="line.131"></a> +<span class="sourceLineNo">132</span> info.getVersionInfo().getVersion()) : ServerMetricsBuilder.of(serverName);<a name="line.132"></a> +<span class="sourceLineNo">133</span> serverManager.checkAndRecordNewServer(serverName, serverMetrics);<a name="line.133"></a> +<span class="sourceLineNo">134</span> }<a name="line.134"></a> +<span class="sourceLineNo">135</span> serverManager.findOutDeadServersAndProcess(deadServersFromPE, liveServersFromWALDir);<a name="line.135"></a> +<span class="sourceLineNo">136</span> }<a name="line.136"></a> +<span class="sourceLineNo">137</span> }<a name="line.137"></a> +<span class="sourceLineNo">138</span><a name="line.138"></a> +<span class="sourceLineNo">139</span> public void stop() {<a name="line.139"></a> +<span class="sourceLineNo">140</span> executor.shutdownNow();<a name="line.140"></a> +<span class="sourceLineNo">141</span> }<a name="line.141"></a> +<span class="sourceLineNo">142</span><a name="line.142"></a> +<span class="sourceLineNo">143</span> private synchronized void refresh() {<a name="line.143"></a> +<span class="sourceLineNo">144</span> List<String> names;<a name="line.144"></a> +<span class="sourceLineNo">145</span> try {<a name="line.145"></a> +<span class="sourceLineNo">146</span> names = ZKUtil.listChildrenAndWatchForNewChildren(watcher, watcher.getZNodePaths().rsZNode);<a name="line.146"></a> +<span class="sourceLineNo">147</span> } catch (KeeperException e) {<a name="line.147"></a> +<span class="sourceLineNo">148</span> // here we need to abort as we failed to set watcher on the rs node which means that we can<a name="line.148"></a> +<span class="sourceLineNo">149</span> // not track the node deleted evetnt any more.<a name="line.149"></a> +<span class="sourceLineNo">150</span> server.abort("Unexpected zk exception getting RS nodes", e);<a name="line.150"></a> +<span class="sourceLineNo">151</span> return;<a name="line.151"></a> +<span class="sourceLineNo">152</span> }<a name="line.152"></a> +<span class="sourceLineNo">153</span> Set<ServerName> servers =<a name="line.153"></a> +<span class="sourceLineNo">154</span> names.stream().map(ServerName::parseServerName).collect(Collectors.toSet());<a name="line.154"></a> +<span class="sourceLineNo">155</span> for (Iterator<ServerName> iter = regionServers.iterator(); iter.hasNext();) {<a name="line.155"></a> +<span class="sourceLineNo">156</span> ServerName sn = iter.next();<a name="line.156"></a> +<span class="sourceLineNo">157</span> if (!servers.contains(sn)) {<a name="line.157"></a> +<span class="sourceLineNo">158</span> LOG.info("RegionServer ephemeral node deleted, processing expiration [{}]", sn);<a name="line.158"></a> +<span class="sourceLineNo">159</span> serverManager.expireServer(sn);<a name="line.159"></a> +<span class="sourceLineNo">160</span> iter.remove();<a name="line.160"></a> +<span class="sourceLineNo">161</span> }<a name="line.161"></a> +<span class="sourceLineNo">162</span> }<a name="line.162"></a> +<span class="sourceLineNo">163</span> // here we do not need to parse the region server info as it is useless now, we only need the<a name="line.163"></a> +<span class="sourceLineNo">164</span> // server name.<a name="line.164"></a> +<span class="sourceLineNo">165</span> boolean newServerAdded = false;<a name="line.165"></a> +<span class="sourceLineNo">166</span> for (ServerName sn : servers) {<a name="line.166"></a> +<span class="sourceLineNo">167</span> if (regionServers.add(sn)) {<a name="line.167"></a> +<span class="sourceLineNo">168</span> newServerAdded = true;<a name="line.168"></a> +<span class="sourceLineNo">169</span> LOG.info("RegionServer ephemeral node created, adding [" + sn + "]");<a name="line.169"></a> +<span class="sourceLineNo">170</span> }<a name="line.170"></a> +<span class="sourceLineNo">171</span> }<a name="line.171"></a> +<span class="sourceLineNo">172</span> if (newServerAdded && server.isInitialized()) {<a name="line.172"></a> +<span class="sourceLineNo">173</span> // Only call the check to move servers if a RegionServer was added to the cluster; in this<a name="line.173"></a> +<span class="sourceLineNo">174</span> // case it could be a server with a new version so it makes sense to run the check.<a name="line.174"></a> +<span class="sourceLineNo">175</span> server.checkIfShouldMoveSystemRegionAsync();<a name="line.175"></a> +<span class="sourceLineNo">176</span> }<a name="line.176"></a> +<span class="sourceLineNo">177</span> }<a name="line.177"></a> +<span class="sourceLineNo">178</span><a name="line.178"></a> +<span class="sourceLineNo">179</span> @Override<a name="line.179"></a> +<span class="sourceLineNo">180</span> public void nodeChildrenChanged(String path) {<a name="line.180"></a> +<span class="sourceLineNo">181</span> if (path.equals(watcher.getZNodePaths().rsZNode) && !server.isAborted() &&<a name="line.181"></a> +<span class="sourceLineNo">182</span> !server.isStopped()) {<a name="line.182"></a> +<span class="sourceLineNo">183</span> executor.execute(this::refresh);<a name="line.183"></a> +<span class="sourceLineNo">184</span> }<a name="line.184"></a> +<span class="sourceLineNo">185</span> }<a name="line.185"></a> +<span class="sourceLineNo">186</span>}<a name="line.186"></a>
http://git-wip-us.apache.org/repos/asf/hbase-site/blob/dbe3a233/devapidocs/src-html/org/apache/hadoop/hbase/master/ServerManager.FlushedSequenceIdFlusher.html ---------------------------------------------------------------------- diff --git a/devapidocs/src-html/org/apache/hadoop/hbase/master/ServerManager.FlushedSequenceIdFlusher.html b/devapidocs/src-html/org/apache/hadoop/hbase/master/ServerManager.FlushedSequenceIdFlusher.html index ba6e4aa..2756f7e 100644 --- a/devapidocs/src-html/org/apache/hadoop/hbase/master/ServerManager.FlushedSequenceIdFlusher.html +++ b/devapidocs/src-html/org/apache/hadoop/hbase/master/ServerManager.FlushedSequenceIdFlusher.html @@ -222,30 +222,30 @@ <span class="sourceLineNo">214</span> /**<a name="line.214"></a> <span class="sourceLineNo">215</span> * Let the server manager know a new regionserver has come online<a name="line.215"></a> <span class="sourceLineNo">216</span> * @param request the startup request<a name="line.216"></a> -<span class="sourceLineNo">217</span> * @param versionNumber the version of the new regionserver<a name="line.217"></a> -<span class="sourceLineNo">218</span> * @param ia the InetAddress from which request is received<a name="line.218"></a> -<span class="sourceLineNo">219</span> * @return The ServerName we know this server as.<a name="line.219"></a> -<span class="sourceLineNo">220</span> * @throws IOException<a name="line.220"></a> -<span class="sourceLineNo">221</span> */<a name="line.221"></a> -<span class="sourceLineNo">222</span> ServerName regionServerStartup(RegionServerStartupRequest request, int versionNumber,<a name="line.222"></a> -<span class="sourceLineNo">223</span> InetAddress ia) throws IOException {<a name="line.223"></a> -<span class="sourceLineNo">224</span> // Test for case where we get a region startup message from a regionserver<a name="line.224"></a> -<span class="sourceLineNo">225</span> // that has been quickly restarted but whose znode expiration handler has<a name="line.225"></a> -<span class="sourceLineNo">226</span> // not yet run, or from a server whose fail we are currently processing.<a name="line.226"></a> -<span class="sourceLineNo">227</span> // Test its host+port combo is present in serverAddressToServerInfo. If it<a name="line.227"></a> -<span class="sourceLineNo">228</span> // is, reject the server and trigger its expiration. The next time it comes<a name="line.228"></a> -<span class="sourceLineNo">229</span> // in, it should have been removed from serverAddressToServerInfo and queued<a name="line.229"></a> -<span class="sourceLineNo">230</span> // for processing by ProcessServerShutdown.<a name="line.230"></a> -<span class="sourceLineNo">231</span><a name="line.231"></a> -<span class="sourceLineNo">232</span> final String hostname = request.hasUseThisHostnameInstead() ?<a name="line.232"></a> -<span class="sourceLineNo">233</span> request.getUseThisHostnameInstead() :ia.getHostName();<a name="line.233"></a> -<span class="sourceLineNo">234</span> ServerName sn = ServerName.valueOf(hostname, request.getPort(),<a name="line.234"></a> -<span class="sourceLineNo">235</span> request.getServerStartCode());<a name="line.235"></a> +<span class="sourceLineNo">217</span> * @param versionNumber the version number of the new regionserver<a name="line.217"></a> +<span class="sourceLineNo">218</span> * @param version the version of the new regionserver, could contain strings like "SNAPSHOT"<a name="line.218"></a> +<span class="sourceLineNo">219</span> * @param ia the InetAddress from which request is received<a name="line.219"></a> +<span class="sourceLineNo">220</span> * @return The ServerName we know this server as.<a name="line.220"></a> +<span class="sourceLineNo">221</span> * @throws IOException<a name="line.221"></a> +<span class="sourceLineNo">222</span> */<a name="line.222"></a> +<span class="sourceLineNo">223</span> ServerName regionServerStartup(RegionServerStartupRequest request, int versionNumber,<a name="line.223"></a> +<span class="sourceLineNo">224</span> String version, InetAddress ia) throws IOException {<a name="line.224"></a> +<span class="sourceLineNo">225</span> // Test for case where we get a region startup message from a regionserver<a name="line.225"></a> +<span class="sourceLineNo">226</span> // that has been quickly restarted but whose znode expiration handler has<a name="line.226"></a> +<span class="sourceLineNo">227</span> // not yet run, or from a server whose fail we are currently processing.<a name="line.227"></a> +<span class="sourceLineNo">228</span> // Test its host+port combo is present in serverAddressToServerInfo. If it<a name="line.228"></a> +<span class="sourceLineNo">229</span> // is, reject the server and trigger its expiration. The next time it comes<a name="line.229"></a> +<span class="sourceLineNo">230</span> // in, it should have been removed from serverAddressToServerInfo and queued<a name="line.230"></a> +<span class="sourceLineNo">231</span> // for processing by ProcessServerShutdown.<a name="line.231"></a> +<span class="sourceLineNo">232</span><a name="line.232"></a> +<span class="sourceLineNo">233</span> final String hostname =<a name="line.233"></a> +<span class="sourceLineNo">234</span> request.hasUseThisHostnameInstead() ? request.getUseThisHostnameInstead() : ia.getHostName();<a name="line.234"></a> +<span class="sourceLineNo">235</span> ServerName sn = ServerName.valueOf(hostname, request.getPort(), request.getServerStartCode());<a name="line.235"></a> <span class="sourceLineNo">236</span> checkClockSkew(sn, request.getServerCurrentTime());<a name="line.236"></a> <span class="sourceLineNo">237</span> checkIsDead(sn, "STARTUP");<a name="line.237"></a> -<span class="sourceLineNo">238</span> if (!checkAndRecordNewServer(sn, ServerMetricsBuilder.of(sn, versionNumber))) {<a name="line.238"></a> -<span class="sourceLineNo">239</span> LOG.warn("THIS SHOULD NOT HAPPEN, RegionServerStartup"<a name="line.239"></a> -<span class="sourceLineNo">240</span> + " could not record the server: " + sn);<a name="line.240"></a> +<span class="sourceLineNo">238</span> if (!checkAndRecordNewServer(sn, ServerMetricsBuilder.of(sn, versionNumber, version))) {<a name="line.238"></a> +<span class="sourceLineNo">239</span> LOG.warn(<a name="line.239"></a> +<span class="sourceLineNo">240</span> "THIS SHOULD NOT HAPPEN, RegionServerStartup" + " could not record the server: " + sn);<a name="line.240"></a> <span class="sourceLineNo">241</span> }<a name="line.241"></a> <span class="sourceLineNo">242</span> return sn;<a name="line.242"></a> <span class="sourceLineNo">243</span> }<a name="line.243"></a> @@ -1029,155 +1029,163 @@ <span class="sourceLineNo">1021</span> /**<a name="line.1021"></a> <span class="sourceLineNo">1022</span> * May return 0 when server is not online.<a name="line.1022"></a> <span class="sourceLineNo">1023</span> */<a name="line.1023"></a> -<span class="sourceLineNo">1024</span> public int getServerVersion(final ServerName serverName) {<a name="line.1024"></a> +<span class="sourceLineNo">1024</span> public int getVersionNumber(ServerName serverName) {<a name="line.1024"></a> <span class="sourceLineNo">1025</span> ServerMetrics serverMetrics = onlineServers.get(serverName);<a name="line.1025"></a> <span class="sourceLineNo">1026</span> return serverMetrics != null ? serverMetrics.getVersionNumber() : 0;<a name="line.1026"></a> <span class="sourceLineNo">1027</span> }<a name="line.1027"></a> <span class="sourceLineNo">1028</span><a name="line.1028"></a> -<span class="sourceLineNo">1029</span> public int getInfoPort(ServerName serverName) {<a name="line.1029"></a> -<span class="sourceLineNo">1030</span> ServerMetrics serverMetrics = onlineServers.get(serverName);<a name="line.1030"></a> -<span class="sourceLineNo">1031</span> return serverMetrics != null ? serverMetrics.getInfoServerPort() : 0;<a name="line.1031"></a> -<span class="sourceLineNo">1032</span> }<a name="line.1032"></a> -<span class="sourceLineNo">1033</span><a name="line.1033"></a> -<span class="sourceLineNo">1034</span> /**<a name="line.1034"></a> -<span class="sourceLineNo">1035</span> * Persist last flushed sequence id of each region to HDFS<a name="line.1035"></a> -<span class="sourceLineNo">1036</span> * @throws IOException if persit to HDFS fails<a name="line.1036"></a> -<span class="sourceLineNo">1037</span> */<a name="line.1037"></a> -<span class="sourceLineNo">1038</span> private void persistRegionLastFlushedSequenceIds() throws IOException {<a name="line.1038"></a> -<span class="sourceLineNo">1039</span> if (isFlushSeqIdPersistInProgress) {<a name="line.1039"></a> -<span class="sourceLineNo">1040</span> return;<a name="line.1040"></a> -<span class="sourceLineNo">1041</span> }<a name="line.1041"></a> -<span class="sourceLineNo">1042</span> isFlushSeqIdPersistInProgress = true;<a name="line.1042"></a> -<span class="sourceLineNo">1043</span> try {<a name="line.1043"></a> -<span class="sourceLineNo">1044</span> Configuration conf = master.getConfiguration();<a name="line.1044"></a> -<span class="sourceLineNo">1045</span> Path rootDir = FSUtils.getRootDir(conf);<a name="line.1045"></a> -<span class="sourceLineNo">1046</span> Path lastFlushedSeqIdPath = new Path(rootDir, LAST_FLUSHED_SEQ_ID_FILE);<a name="line.1046"></a> -<span class="sourceLineNo">1047</span> FileSystem fs = FileSystem.get(conf);<a name="line.1047"></a> -<span class="sourceLineNo">1048</span> if (fs.exists(lastFlushedSeqIdPath)) {<a name="line.1048"></a> -<span class="sourceLineNo">1049</span> LOG.info("Rewriting .lastflushedseqids file at: "<a name="line.1049"></a> -<span class="sourceLineNo">1050</span> + lastFlushedSeqIdPath);<a name="line.1050"></a> -<span class="sourceLineNo">1051</span> if (!fs.delete(lastFlushedSeqIdPath, false)) {<a name="line.1051"></a> -<span class="sourceLineNo">1052</span> throw new IOException("Unable to remove existing "<a name="line.1052"></a> -<span class="sourceLineNo">1053</span> + lastFlushedSeqIdPath);<a name="line.1053"></a> -<span class="sourceLineNo">1054</span> }<a name="line.1054"></a> -<span class="sourceLineNo">1055</span> } else {<a name="line.1055"></a> -<span class="sourceLineNo">1056</span> LOG.info("Writing .lastflushedseqids file at: " + lastFlushedSeqIdPath);<a name="line.1056"></a> -<span class="sourceLineNo">1057</span> }<a name="line.1057"></a> -<span class="sourceLineNo">1058</span> FSDataOutputStream out = fs.create(lastFlushedSeqIdPath);<a name="line.1058"></a> -<span class="sourceLineNo">1059</span> FlushedSequenceId.Builder flushedSequenceIdBuilder =<a name="line.1059"></a> -<span class="sourceLineNo">1060</span> FlushedSequenceId.newBuilder();<a name="line.1060"></a> -<span class="sourceLineNo">1061</span> try {<a name="line.1061"></a> -<span class="sourceLineNo">1062</span> for (Entry<byte[], Long> entry : flushedSequenceIdByRegion.entrySet()) {<a name="line.1062"></a> -<span class="sourceLineNo">1063</span> FlushedRegionSequenceId.Builder flushedRegionSequenceIdBuilder =<a name="line.1063"></a> -<span class="sourceLineNo">1064</span> FlushedRegionSequenceId.newBuilder();<a name="line.1064"></a> -<span class="sourceLineNo">1065</span> flushedRegionSequenceIdBuilder.setRegionEncodedName(<a name="line.1065"></a> -<span class="sourceLineNo">1066</span> ByteString.copyFrom(entry.getKey()));<a name="line.1066"></a> -<span class="sourceLineNo">1067</span> flushedRegionSequenceIdBuilder.setSeqId(entry.getValue());<a name="line.1067"></a> -<span class="sourceLineNo">1068</span> ConcurrentNavigableMap<byte[], Long> storeSeqIds =<a name="line.1068"></a> -<span class="sourceLineNo">1069</span> storeFlushedSequenceIdsByRegion.get(entry.getKey());<a name="line.1069"></a> -<span class="sourceLineNo">1070</span> if (storeSeqIds != null) {<a name="line.1070"></a> -<span class="sourceLineNo">1071</span> for (Entry<byte[], Long> store : storeSeqIds.entrySet()) {<a name="line.1071"></a> -<span class="sourceLineNo">1072</span> FlushedStoreSequenceId.Builder flushedStoreSequenceIdBuilder =<a name="line.1072"></a> -<span class="sourceLineNo">1073</span> FlushedStoreSequenceId.newBuilder();<a name="line.1073"></a> -<span class="sourceLineNo">1074</span> flushedStoreSequenceIdBuilder.setFamily(ByteString.copyFrom(store.getKey()));<a name="line.1074"></a> -<span class="sourceLineNo">1075</span> flushedStoreSequenceIdBuilder.setSeqId(store.getValue());<a name="line.1075"></a> -<span class="sourceLineNo">1076</span> flushedRegionSequenceIdBuilder.addStores(flushedStoreSequenceIdBuilder);<a name="line.1076"></a> -<span class="sourceLineNo">1077</span> }<a name="line.1077"></a> -<span class="sourceLineNo">1078</span> }<a name="line.1078"></a> -<span class="sourceLineNo">1079</span> flushedSequenceIdBuilder.addRegionSequenceId(flushedRegionSequenceIdBuilder);<a name="line.1079"></a> -<span class="sourceLineNo">1080</span> }<a name="line.1080"></a> -<span class="sourceLineNo">1081</span> flushedSequenceIdBuilder.build().writeDelimitedTo(out);<a name="line.1081"></a> -<span class="sourceLineNo">1082</span> } finally {<a name="line.1082"></a> -<span class="sourceLineNo">1083</span> if (out != null) {<a name="line.1083"></a> -<span class="sourceLineNo">1084</span> out.close();<a name="line.1084"></a> -<span class="sourceLineNo">1085</span> }<a name="line.1085"></a> -<span class="sourceLineNo">1086</span> }<a name="line.1086"></a> -<span class="sourceLineNo">1087</span> } finally {<a name="line.1087"></a> -<span class="sourceLineNo">1088</span> isFlushSeqIdPersistInProgress = false;<a name="line.1088"></a> -<span class="sourceLineNo">1089</span> }<a name="line.1089"></a> -<span class="sourceLineNo">1090</span> }<a name="line.1090"></a> -<span class="sourceLineNo">1091</span><a name="line.1091"></a> -<span class="sourceLineNo">1092</span> /**<a name="line.1092"></a> -<span class="sourceLineNo">1093</span> * Load last flushed sequence id of each region from HDFS, if persisted<a name="line.1093"></a> -<span class="sourceLineNo">1094</span> */<a name="line.1094"></a> -<span class="sourceLineNo">1095</span> public void loadLastFlushedSequenceIds() throws IOException {<a name="line.1095"></a> -<span class="sourceLineNo">1096</span> if (!persistFlushedSequenceId) {<a name="line.1096"></a> -<span class="sourceLineNo">1097</span> return;<a name="line.1097"></a> -<span class="sourceLineNo">1098</span> }<a name="line.1098"></a> -<span class="sourceLineNo">1099</span> Configuration conf = master.getConfiguration();<a name="line.1099"></a> -<span class="sourceLineNo">1100</span> Path rootDir = FSUtils.getRootDir(conf);<a name="line.1100"></a> -<span class="sourceLineNo">1101</span> Path lastFlushedSeqIdPath = new Path(rootDir, LAST_FLUSHED_SEQ_ID_FILE);<a name="line.1101"></a> -<span class="sourceLineNo">1102</span> FileSystem fs = FileSystem.get(conf);<a name="line.1102"></a> -<span class="sourceLineNo">1103</span> if (!fs.exists(lastFlushedSeqIdPath)) {<a name="line.1103"></a> -<span class="sourceLineNo">1104</span> LOG.info("No .lastflushedseqids found at" + lastFlushedSeqIdPath<a name="line.1104"></a> -<span class="sourceLineNo">1105</span> + " will record last flushed sequence id"<a name="line.1105"></a> -<span class="sourceLineNo">1106</span> + " for regions by regionserver report all over again");<a name="line.1106"></a> -<span class="sourceLineNo">1107</span> return;<a name="line.1107"></a> -<span class="sourceLineNo">1108</span> } else {<a name="line.1108"></a> -<span class="sourceLineNo">1109</span> LOG.info("begin to load .lastflushedseqids at " + lastFlushedSeqIdPath);<a name="line.1109"></a> -<span class="sourceLineNo">1110</span> }<a name="line.1110"></a> -<span class="sourceLineNo">1111</span> FSDataInputStream in = fs.open(lastFlushedSeqIdPath);<a name="line.1111"></a> -<span class="sourceLineNo">1112</span> try {<a name="line.1112"></a> -<span class="sourceLineNo">1113</span> FlushedSequenceId flushedSequenceId =<a name="line.1113"></a> -<span class="sourceLineNo">1114</span> FlushedSequenceId.parseDelimitedFrom(in);<a name="line.1114"></a> -<span class="sourceLineNo">1115</span> for (FlushedRegionSequenceId flushedRegionSequenceId : flushedSequenceId<a name="line.1115"></a> -<span class="sourceLineNo">1116</span> .getRegionSequenceIdList()) {<a name="line.1116"></a> -<span class="sourceLineNo">1117</span> byte[] encodedRegionName = flushedRegionSequenceId<a name="line.1117"></a> -<span class="sourceLineNo">1118</span> .getRegionEncodedName().toByteArray();<a name="line.1118"></a> -<span class="sourceLineNo">1119</span> flushedSequenceIdByRegion<a name="line.1119"></a> -<span class="sourceLineNo">1120</span> .putIfAbsent(encodedRegionName, flushedRegionSequenceId.getSeqId());<a name="line.1120"></a> -<span class="sourceLineNo">1121</span> if (flushedRegionSequenceId.getStoresList() != null<a name="line.1121"></a> -<span class="sourceLineNo">1122</span> && flushedRegionSequenceId.getStoresList().size() != 0) {<a name="line.1122"></a> -<span class="sourceLineNo">1123</span> ConcurrentNavigableMap<byte[], Long> storeFlushedSequenceId =<a name="line.1123"></a> -<span class="sourceLineNo">1124</span> computeIfAbsent(storeFlushedSequenceIdsByRegion, encodedRegionName,<a name="line.1124"></a> -<span class="sourceLineNo">1125</span> () -> new ConcurrentSkipListMap<>(Bytes.BYTES_COMPARATOR));<a name="line.1125"></a> -<span class="sourceLineNo">1126</span> for (FlushedStoreSequenceId flushedStoreSequenceId : flushedRegionSequenceId<a name="line.1126"></a> -<span class="sourceLineNo">1127</span> .getStoresList()) {<a name="line.1127"></a> -<span class="sourceLineNo">1128</span> storeFlushedSequenceId<a name="line.1128"></a> -<span class="sourceLineNo">1129</span> .put(flushedStoreSequenceId.getFamily().toByteArray(),<a name="line.1129"></a> -<span class="sourceLineNo">1130</span> flushedStoreSequenceId.getSeqId());<a name="line.1130"></a> -<span class="sourceLineNo">1131</span> }<a name="line.1131"></a> -<span class="sourceLineNo">1132</span> }<a name="line.1132"></a> -<span class="sourceLineNo">1133</span> }<a name="line.1133"></a> -<span class="sourceLineNo">1134</span> } finally {<a name="line.1134"></a> -<span class="sourceLineNo">1135</span> in.close();<a name="line.1135"></a> -<span class="sourceLineNo">1136</span> }<a name="line.1136"></a> -<span class="sourceLineNo">1137</span> }<a name="line.1137"></a> -<span class="sourceLineNo">1138</span><a name="line.1138"></a> -<span class="sourceLineNo">1139</span> /**<a name="line.1139"></a> -<span class="sourceLineNo">1140</span> * Regions may have been removed between latest persist of FlushedSequenceIds<a name="line.1140"></a> -<span class="sourceLineNo">1141</span> * and master abort. So after loading FlushedSequenceIds from file, and after<a name="line.1141"></a> -<span class="sourceLineNo">1142</span> * meta loaded, we need to remove the deleted region according to RegionStates.<a name="line.1142"></a> -<span class="sourceLineNo">1143</span> */<a name="line.1143"></a> -<span class="sourceLineNo">1144</span> public void removeDeletedRegionFromLoadedFlushedSequenceIds() {<a name="line.1144"></a> -<span class="sourceLineNo">1145</span> RegionStates regionStates = master.getAssignmentManager().getRegionStates();<a name="line.1145"></a> -<span class="sourceLineNo">1146</span> Iterator<byte[]> it = flushedSequenceIdByRegion.keySet().iterator();<a name="line.1146"></a> -<span class="sourceLineNo">1147</span> while(it.hasNext()) {<a name="line.1147"></a> -<span class="sourceLineNo">1148</span> byte[] regionEncodedName = it.next();<a name="line.1148"></a> -<span class="sourceLineNo">1149</span> if (regionStates.getRegionState(Bytes.toStringBinary(regionEncodedName)) == null) {<a name="line.1149"></a> -<span class="sourceLineNo">1150</span> it.remove();<a name="line.1150"></a> -<span class="sourceLineNo">1151</span> storeFlushedSequenceIdsByRegion.remove(regionEncodedName);<a name="line.1151"></a> -<span class="sourceLineNo">1152</span> }<a name="line.1152"></a> -<span class="sourceLineNo">1153</span> }<a name="line.1153"></a> -<span class="sourceLineNo">1154</span> }<a name="line.1154"></a> -<span class="sourceLineNo">1155</span><a name="line.1155"></a> -<span class="sourceLineNo">1156</span> private class FlushedSequenceIdFlusher extends ScheduledChore {<a name="line.1156"></a> -<span class="sourceLineNo">1157</span><a name="line.1157"></a> -<span class="sourceLineNo">1158</span> public FlushedSequenceIdFlusher(String name, int p) {<a name="line.1158"></a> -<span class="sourceLineNo">1159</span> super(name, master, p, 60 * 1000); //delay one minute before first execute<a name="line.1159"></a> -<span class="sourceLineNo">1160</span> }<a name="line.1160"></a> -<span class="sourceLineNo">1161</span><a name="line.1161"></a> -<span class="sourceLineNo">1162</span> @Override<a name="line.1162"></a> -<span class="sourceLineNo">1163</span> protected void chore() {<a name="line.1163"></a> -<span class="sourceLineNo">1164</span> try {<a name="line.1164"></a> -<span class="sourceLineNo">1165</span> persistRegionLastFlushedSequenceIds();<a name="line.1165"></a> -<span class="sourceLineNo">1166</span> } catch (IOException e) {<a name="line.1166"></a> -<span class="sourceLineNo">1167</span> LOG.debug("Failed to persist last flushed sequence id of regions"<a name="line.1167"></a> -<span class="sourceLineNo">1168</span> + " to file system", e);<a name="line.1168"></a> -<span class="sourceLineNo">1169</span> }<a name="line.1169"></a> -<span class="sourceLineNo">1170</span> }<a name="line.1170"></a> -<span class="sourceLineNo">1171</span> }<a name="line.1171"></a> -<span class="sourceLineNo">1172</span>}<a name="line.1172"></a> +<span class="sourceLineNo">1029</span> /**<a name="line.1029"></a> +<span class="sourceLineNo">1030</span> * May return "0.0.0" when server is not online<a name="line.1030"></a> +<span class="sourceLineNo">1031</span> */<a name="line.1031"></a> +<span class="sourceLineNo">1032</span> public String getVersion(ServerName serverName) {<a name="line.1032"></a> +<span class="sourceLineNo">1033</span> ServerMetrics serverMetrics = onlineServers.get(serverName);<a name="line.1033"></a> +<span class="sourceLineNo">1034</span> return serverMetrics != null ? serverMetrics.getVersion() : "0.0.0";<a name="line.1034"></a> +<span class="sourceLineNo">1035</span> }<a name="line.1035"></a> +<span class="sourceLineNo">1036</span><a name="line.1036"></a> +<span class="sourceLineNo">1037</span> public int getInfoPort(ServerName serverName) {<a name="line.1037"></a> +<span class="sourceLineNo">1038</span> ServerMetrics serverMetrics = onlineServers.get(serverName);<a name="line.1038"></a> +<span class="sourceLineNo">1039</span> return serverMetrics != null ? serverMetrics.getInfoServerPort() : 0;<a name="line.1039"></a> +<span class="sourceLineNo">1040</span> }<a name="line.1040"></a> +<span class="sourceLineNo">1041</span><a name="line.1041"></a> +<span class="sourceLineNo">1042</span> /**<a name="line.1042"></a> +<span class="sourceLineNo">1043</span> * Persist last flushed sequence id of each region to HDFS<a name="line.1043"></a> +<span class="sourceLineNo">1044</span> * @throws IOException if persit to HDFS fails<a name="line.1044"></a> +<span class="sourceLineNo">1045</span> */<a name="line.1045"></a> +<span class="sourceLineNo">1046</span> private void persistRegionLastFlushedSequenceIds() throws IOException {<a name="line.1046"></a> +<span class="sourceLineNo">1047</span> if (isFlushSeqIdPersistInProgress) {<a name="line.1047"></a> +<span class="sourceLineNo">1048</span> return;<a name="line.1048"></a> +<span class="sourceLineNo">1049</span> }<a name="line.1049"></a> +<span class="sourceLineNo">1050</span> isFlushSeqIdPersistInProgress = true;<a name="line.1050"></a> +<span class="sourceLineNo">1051</span> try {<a name="line.1051"></a> +<span class="sourceLineNo">1052</span> Configuration conf = master.getConfiguration();<a name="line.1052"></a> +<span class="sourceLineNo">1053</span> Path rootDir = FSUtils.getRootDir(conf);<a name="line.1053"></a> +<span class="sourceLineNo">1054</span> Path lastFlushedSeqIdPath = new Path(rootDir, LAST_FLUSHED_SEQ_ID_FILE);<a name="line.1054"></a> +<span class="sourceLineNo">1055</span> FileSystem fs = FileSystem.get(conf);<a name="line.1055"></a> +<span class="sourceLineNo">1056</span> if (fs.exists(lastFlushedSeqIdPath)) {<a name="line.1056"></a> +<span class="sourceLineNo">1057</span> LOG.info("Rewriting .lastflushedseqids file at: "<a name="line.1057"></a> +<span class="sourceLineNo">1058</span> + lastFlushedSeqIdPath);<a name="line.1058"></a> +<span class="sourceLineNo">1059</span> if (!fs.delete(lastFlushedSeqIdPath, false)) {<a name="line.1059"></a> +<span class="sourceLineNo">1060</span> throw new IOException("Unable to remove existing "<a name="line.1060"></a> +<span class="sourceLineNo">1061</span> + lastFlushedSeqIdPath);<a name="line.1061"></a> +<span class="sourceLineNo">1062</span> }<a name="line.1062"></a> +<span class="sourceLineNo">1063</span> } else {<a name="line.1063"></a> +<span class="sourceLineNo">1064</span> LOG.info("Writing .lastflushedseqids file at: " + lastFlushedSeqIdPath);<a name="line.1064"></a> +<span class="sourceLineNo">1065</span> }<a name="line.1065"></a> +<span class="sourceLineNo">1066</span> FSDataOutputStream out = fs.create(lastFlushedSeqIdPath);<a name="line.1066"></a> +<span class="sourceLineNo">1067</span> FlushedSequenceId.Builder flushedSequenceIdBuilder =<a name="line.1067"></a> +<span class="sourceLineNo">1068</span> FlushedSequenceId.newBuilder();<a name="line.1068"></a> +<span class="sourceLineNo">1069</span> try {<a name="line.1069"></a> +<span class="sourceLineNo">1070</span> for (Entry<byte[], Long> entry : flushedSequenceIdByRegion.entrySet()) {<a name="line.1070"></a> +<span class="sourceLineNo">1071</span> FlushedRegionSequenceId.Builder flushedRegionSequenceIdBuilder =<a name="line.1071"></a> +<span class="sourceLineNo">1072</span> FlushedRegionSequenceId.newBuilder();<a name="line.1072"></a> +<span class="sourceLineNo">1073</span> flushedRegionSequenceIdBuilder.setRegionEncodedName(<a name="line.1073"></a> +<span class="sourceLineNo">1074</span> ByteString.copyFrom(entry.getKey()));<a name="line.1074"></a> +<span class="sourceLineNo">1075</span> flushedRegionSequenceIdBuilder.setSeqId(entry.getValue());<a name="line.1075"></a> +<span class="sourceLineNo">1076</span> ConcurrentNavigableMap<byte[], Long> storeSeqIds =<a name="line.1076"></a> +<span class="sourceLineNo">1077</span> storeFlushedSequenceIdsByRegion.get(entry.getKey());<a name="line.1077"></a> +<span class="sourceLineNo">1078</span> if (storeSeqIds != null) {<a name="line.1078"></a> +<span class="sourceLineNo">1079</span> for (Entry<byte[], Long> store : storeSeqIds.entrySet()) {<a name="line.1079"></a> +<span class="sourceLineNo">1080</span> FlushedStoreSequenceId.Builder flushedStoreSequenceIdBuilder =<a name="line.1080"></a> +<span class="sourceLineNo">1081</span> FlushedStoreSequenceId.newBuilder();<a name="line.1081"></a> +<span class="sourceLineNo">1082</span> flushedStoreSequenceIdBuilder.setFamily(ByteString.copyFrom(store.getKey()));<a name="line.1082"></a> +<span class="sourceLineNo">1083</span> flushedStoreSequenceIdBuilder.setSeqId(store.getValue());<a name="line.1083"></a> +<span class="sourceLineNo">1084</span> flushedRegionSequenceIdBuilder.addStores(flushedStoreSequenceIdBuilder);<a name="line.1084"></a> +<span class="sourceLineNo">1085</span> }<a name="line.1085"></a> +<span class="sourceLineNo">1086</span> }<a name="line.1086"></a> +<span class="sourceLineNo">1087</span> flushedSequenceIdBuilder.addRegionSequenceId(flushedRegionSequenceIdBuilder);<a name="line.1087"></a> +<span class="sourceLineNo">1088</span> }<a name="line.1088"></a> +<span class="sourceLineNo">1089</span> flushedSequenceIdBuilder.build().writeDelimitedTo(out);<a name="line.1089"></a> +<span class="sourceLineNo">1090</span> } finally {<a name="line.1090"></a> +<span class="sourceLineNo">1091</span> if (out != null) {<a name="line.1091"></a> +<span class="sourceLineNo">1092</span> out.close();<a name="line.1092"></a> +<span class="sourceLineNo">1093</span> }<a name="line.1093"></a> +<span class="sourceLineNo">1094</span> }<a name="line.1094"></a> +<span class="sourceLineNo">1095</span> } finally {<a name="line.1095"></a> +<span class="sourceLineNo">1096</span> isFlushSeqIdPersistInProgress = false;<a name="line.1096"></a> +<span class="sourceLineNo">1097</span> }<a name="line.1097"></a> +<span class="sourceLineNo">1098</span> }<a name="line.1098"></a> +<span class="sourceLineNo">1099</span><a name="line.1099"></a> +<span class="sourceLineNo">1100</span> /**<a name="line.1100"></a> +<span class="sourceLineNo">1101</span> * Load last flushed sequence id of each region from HDFS, if persisted<a name="line.1101"></a> +<span class="sourceLineNo">1102</span> */<a name="line.1102"></a> +<span class="sourceLineNo">1103</span> public void loadLastFlushedSequenceIds() throws IOException {<a name="line.1103"></a> +<span class="sourceLineNo">1104</span> if (!persistFlushedSequenceId) {<a name="line.1104"></a> +<span class="sourceLineNo">1105</span> return;<a name="line.1105"></a> +<span class="sourceLineNo">1106</span> }<a name="line.1106"></a> +<span class="sourceLineNo">1107</span> Configuration conf = master.getConfiguration();<a name="line.1107"></a> +<span class="sourceLineNo">1108</span> Path rootDir = FSUtils.getRootDir(conf);<a name="line.1108"></a> +<span class="sourceLineNo">1109</span> Path lastFlushedSeqIdPath = new Path(rootDir, LAST_FLUSHED_SEQ_ID_FILE);<a name="line.1109"></a> +<span class="sourceLineNo">1110</span> FileSystem fs = FileSystem.get(conf);<a name="line.1110"></a> +<span class="sourceLineNo">1111</span> if (!fs.exists(lastFlushedSeqIdPath)) {<a name="line.1111"></a> +<span class="sourceLineNo">1112</span> LOG.info("No .lastflushedseqids found at" + lastFlushedSeqIdPath<a name="line.1112"></a> +<span class="sourceLineNo">1113</span> + " will record last flushed sequence id"<a name="line.1113"></a> +<span class="sourceLineNo">1114</span> + " for regions by regionserver report all over again");<a name="line.1114"></a> +<span class="sourceLineNo">1115</span> return;<a name="line.1115"></a> +<span class="sourceLineNo">1116</span> } else {<a name="line.1116"></a> +<span class="sourceLineNo">1117</span> LOG.info("begin to load .lastflushedseqids at " + lastFlushedSeqIdPath);<a name="line.1117"></a> +<span class="sourceLineNo">1118</span> }<a name="line.1118"></a> +<span class="sourceLineNo">1119</span> FSDataInputStream in = fs.open(lastFlushedSeqIdPath);<a name="line.1119"></a> +<span class="sourceLineNo">1120</span> try {<a name="line.1120"></a> +<span class="sourceLineNo">1121</span> FlushedSequenceId flushedSequenceId =<a name="line.1121"></a> +<span class="sourceLineNo">1122</span> FlushedSequenceId.parseDelimitedFrom(in);<a name="line.1122"></a> +<span class="sourceLineNo">1123</span> for (FlushedRegionSequenceId flushedRegionSequenceId : flushedSequenceId<a name="line.1123"></a> +<span class="sourceLineNo">1124</span> .getRegionSequenceIdList()) {<a name="line.1124"></a> +<span class="sourceLineNo">1125</span> byte[] encodedRegionName = flushedRegionSequenceId<a name="line.1125"></a> +<span class="sourceLineNo">1126</span> .getRegionEncodedName().toByteArray();<a name="line.1126"></a> +<span class="sourceLineNo">1127</span> flushedSequenceIdByRegion<a name="line.1127"></a> +<span class="sourceLineNo">1128</span> .putIfAbsent(encodedRegionName, flushedRegionSequenceId.getSeqId());<a name="line.1128"></a> +<span class="sourceLineNo">1129</span> if (flushedRegionSequenceId.getStoresList() != null<a name="line.1129"></a> +<span class="sourceLineNo">1130</span> && flushedRegionSequenceId.getStoresList().size() != 0) {<a name="line.1130"></a> +<span class="sourceLineNo">1131</span> ConcurrentNavigableMap<byte[], Long> storeFlushedSequenceId =<a name="line.1131"></a> +<span class="sourceLineNo">1132</span> computeIfAbsent(storeFlushedSequenceIdsByRegion, encodedRegionName,<a name="line.1132"></a> +<span class="sourceLineNo">1133</span> () -> new ConcurrentSkipListMap<>(Bytes.BYTES_COMPARATOR));<a name="line.1133"></a> +<span class="sourceLineNo">1134</span> for (FlushedStoreSequenceId flushedStoreSequenceId : flushedRegionSequenceId<a name="line.1134"></a> +<span class="sourceLineNo">1135</span> .getStoresList()) {<a name="line.1135"></a> +<span class="sourceLineNo">1136</span> storeFlushedSequenceId<a name="line.1136"></a> +<span class="sourceLineNo">1137</span> .put(flushedStoreSequenceId.getFamily().toByteArray(),<a name="line.1137"></a> +<span class="sourceLineNo">1138</span> flushedStoreSequenceId.getSeqId());<a name="line.1138"></a> +<span class="sourceLineNo">1139</span> }<a name="line.1139"></a> +<span class="sourceLineNo">1140</span> }<a name="line.1140"></a> +<span class="sourceLineNo">1141</span> }<a name="line.1141"></a> +<span class="sourceLineNo">1142</span> } finally {<a name="line.1142"></a> +<span class="sourceLineNo">1143</span> in.close();<a name="line.1143"></a> +<span class="sourceLineNo">1144</span> }<a name="line.1144"></a> +<span class="sourceLineNo">1145</span> }<a name="line.1145"></a> +<span class="sourceLineNo">1146</span><a name="line.1146"></a> +<span class="sourceLineNo">1147</span> /**<a name="line.1147"></a> +<span class="sourceLineNo">1148</span> * Regions may have been removed between latest persist of FlushedSequenceIds<a name="line.1148"></a> +<span class="sourceLineNo">1149</span> * and master abort. So after loading FlushedSequenceIds from file, and after<a name="line.1149"></a> +<span class="sourceLineNo">1150</span> * meta loaded, we need to remove the deleted region according to RegionStates.<a name="line.1150"></a> +<span class="sourceLineNo">1151</span> */<a name="line.1151"></a> +<span class="sourceLineNo">1152</span> public void removeDeletedRegionFromLoadedFlushedSequenceIds() {<a name="line.1152"></a> +<span class="sourceLineNo">1153</span> RegionStates regionStates = master.getAssignmentManager().getRegionStates();<a name="line.1153"></a> +<span class="sourceLineNo">1154</span> Iterator<byte[]> it = flushedSequenceIdByRegion.keySet().iterator();<a name="line.1154"></a> +<span class="sourceLineNo">1155</span> while(it.hasNext()) {<a name="line.1155"></a> +<span class="sourceLineNo">1156</span> byte[] regionEncodedName = it.next();<a name="line.1156"></a> +<span class="sourceLineNo">1157</span> if (regionStates.getRegionState(Bytes.toStringBinary(regionEncodedName)) == null) {<a name="line.1157"></a> +<span class="sourceLineNo">1158</span> it.remove();<a name="line.1158"></a> +<span class="sourceLineNo">1159</span> storeFlushedSequenceIdsByRegion.remove(regionEncodedName);<a name="line.1159"></a> +<span class="sourceLineNo">1160</span> }<a name="line.1160"></a> +<span class="sourceLineNo">1161</span> }<a name="line.1161"></a> +<span class="sourceLineNo">1162</span> }<a name="line.1162"></a> +<span class="sourceLineNo">1163</span><a name="line.1163"></a> +<span class="sourceLineNo">1164</span> private class FlushedSequenceIdFlusher extends ScheduledChore {<a name="line.1164"></a> +<span class="sourceLineNo">1165</span><a name="line.1165"></a> +<span class="sourceLineNo">1166</span> public FlushedSequenceIdFlusher(String name, int p) {<a name="line.1166"></a> +<span class="sourceLineNo">1167</span> super(name, master, p, 60 * 1000); //delay one minute before first execute<a name="line.1167"></a> +<span class="sourceLineNo">1168</span> }<a name="line.1168"></a> +<span class="sourceLineNo">1169</span><a name="line.1169"></a> +<span class="sourceLineNo">1170</span> @Override<a name="line.1170"></a> +<span class="sourceLineNo">1171</span> protected void chore() {<a name="line.1171"></a> +<span class="sourceLineNo">1172</span> try {<a name="line.1172"></a> +<span class="sourceLineNo">1173</span> persistRegionLastFlushedSequenceIds();<a name="line.1173"></a> +<span class="sourceLineNo">1174</span> } catch (IOException e) {<a name="line.1174"></a> +<span class="sourceLineNo">1175</span> LOG.debug("Failed to persist last flushed sequence id of regions"<a name="line.1175"></a> +<span class="sourceLineNo">1176</span> + " to file system", e);<a name="line.1176"></a> +<span class="sourceLineNo">1177</span> }<a name="line.1177"></a> +<span class="sourceLineNo">1178</span> }<a name="line.1178"></a> +<span class="sourceLineNo">1179</span> }<a name="line.1179"></a> +<span class="sourceLineNo">1180</span>}<a name="line.1180"></a> http://git-wip-us.apache.org/repos/asf/hbase-site/blob/dbe3a233/devapidocs/src-html/org/apache/hadoop/hbase/master/ServerManager.html ---------------------------------------------------------------------- diff --git a/devapidocs/src-html/org/apache/hadoop/hbase/master/ServerManager.html b/devapidocs/src-html/org/apache/hadoop/hbase/master/ServerManager.html index ba6e4aa..2756f7e 100644 --- a/devapidocs/src-html/org/apache/hadoop/hbase/master/ServerManager.html +++ b/devapidocs/src-html/org/apache/hadoop/hbase/master/ServerManager.html @@ -222,30 +222,30 @@ <span class="sourceLineNo">214</span> /**<a name="line.214"></a> <span class="sourceLineNo">215</span> * Let the server manager know a new regionserver has come online<a name="line.215"></a> <span class="sourceLineNo">216</span> * @param request the startup request<a name="line.216"></a> -<span class="sourceLineNo">217</span> * @param versionNumber the version of the new regionserver<a name="line.217"></a> -<span class="sourceLineNo">218</span> * @param ia the InetAddress from which request is received<a name="line.218"></a> -<span class="sourceLineNo">219</span> * @return The ServerName we know this server as.<a name="line.219"></a> -<span class="sourceLineNo">220</span> * @throws IOException<a name="line.220"></a> -<span class="sourceLineNo">221</span> */<a name="line.221"></a> -<span class="sourceLineNo">222</span> ServerName regionServerStartup(RegionServerStartupRequest request, int versionNumber,<a name="line.222"></a> -<span class="sourceLineNo">223</span> InetAddress ia) throws IOException {<a name="line.223"></a> -<span class="sourceLineNo">224</span> // Test for case where we get a region startup message from a regionserver<a name="line.224"></a> -<span class="sourceLineNo">225</span> // that has been quickly restarted but whose znode expiration handler has<a name="line.225"></a> -<span class="sourceLineNo">226</span> // not yet run, or from a server whose fail we are currently processing.<a name="line.226"></a> -<span class="sourceLineNo">227</span> // Test its host+port combo is present in serverAddressToServerInfo. If it<a name="line.227"></a> -<span class="sourceLineNo">228</span> // is, reject the server and trigger its expiration. The next time it comes<a name="line.228"></a> -<span class="sourceLineNo">229</span> // in, it should have been removed from serverAddressToServerInfo and queued<a name="line.229"></a> -<span class="sourceLineNo">230</span> // for processing by ProcessServerShutdown.<a name="line.230"></a> -<span class="sourceLineNo">231</span><a name="line.231"></a> -<span class="sourceLineNo">232</span> final String hostname = request.hasUseThisHostnameInstead() ?<a name="line.232"></a> -<span class="sourceLineNo">233</span> request.getUseThisHostnameInstead() :ia.getHostName();<a name="line.233"></a> -<span class="sourceLineNo">234</span> ServerName sn = ServerName.valueOf(hostname, request.getPort(),<a name="line.234"></a> -<span class="sourceLineNo">235</span> request.getServerStartCode());<a name="line.235"></a> +<span class="sourceLineNo">217</span> * @param versionNumber the version number of the new regionserver<a name="line.217"></a> +<span class="sourceLineNo">218</span> * @param version the version of the new regionserver, could contain strings like "SNAPSHOT"<a name="line.218"></a> +<span class="sourceLineNo">219</span> * @param ia the InetAddress from which request is received<a name="line.219"></a> +<span class="sourceLineNo">220</span> * @return The ServerName we know this server as.<a name="line.220"></a> +<span class="sourceLineNo">221</span> * @throws IOException<a name="line.221"></a> +<span class="sourceLineNo">222</span> */<a name="line.222"></a> +<span class="sourceLineNo">223</span> ServerName regionServerStartup(RegionServerStartupRequest request, int versionNumber,<a name="line.223"></a> +<span class="sourceLineNo">224</span> String version, InetAddress ia) throws IOException {<a name="line.224"></a> +<span class="sourceLineNo">225</span> // Test for case where we get a region startup message from a regionserver<a name="line.225"></a> +<span class="sourceLineNo">226</span> // that has been quickly restarted but whose znode expiration handler has<a name="line.226"></a> +<span class="sourceLineNo">227</span> // not yet run, or from a server whose fail we are currently processing.<a name="line.227"></a> +<span class="sourceLineNo">228</span> // Test its host+port combo is present in serverAddressToServerInfo. If it<a name="line.228"></a> +<span class="sourceLineNo">229</span> // is, reject the server and trigger its expiration. The next time it comes<a name="line.229"></a> +<span class="sourceLineNo">230</span> // in, it should have been removed from serverAddressToServerInfo and queued<a name="line.230"></a> +<span class="sourceLineNo">231</span> // for processing by ProcessServerShutdown.<a name="line.231"></a> +<span class="sourceLineNo">232</span><a name="line.232"></a> +<span class="sourceLineNo">233</span> final String hostname =<a name="line.233"></a> +<span class="sourceLineNo">234</span> request.hasUseThisHostnameInstead() ? request.getUseThisHostnameInstead() : ia.getHostName();<a name="line.234"></a> +<span class="sourceLineNo">235</span> ServerName sn = ServerName.valueOf(hostname, request.getPort(), request.getServerStartCode());<a name="line.235"></a> <span class="sourceLineNo">236</span> checkClockSkew(sn, request.getServerCurrentTime());<a name="line.236"></a> <span class="sourceLineNo">237</span> checkIsDead(sn, "STARTUP");<a name="line.237"></a> -<span class="sourceLineNo">238</span> if (!checkAndRecordNewServer(sn, ServerMetricsBuilder.of(sn, versionNumber))) {<a name="line.238"></a> -<span class="sourceLineNo">239</span> LOG.warn("THIS SHOULD NOT HAPPEN, RegionServerStartup"<a name="line.239"></a> -<span class="sourceLineNo">240</span> + " could not record the server: " + sn);<a name="line.240"></a> +<span class="sourceLineNo">238</span> if (!checkAndRecordNewServer(sn, ServerMetricsBuilder.of(sn, versionNumber, version))) {<a name="line.238"></a> +<span class="sourceLineNo">239</span> LOG.warn(<a name="line.239"></a> +<span class="sourceLineNo">240</span> "THIS SHOULD NOT HAPPEN, RegionServerStartup" + " could not record the server: " + sn);<a name="line.240"></a> <span class="sourceLineNo">241</span> }<a name="line.241"></a> <span class="sourceLineNo">242</span> return sn;<a name="line.242"></a> <span class="sourceLineNo">243</span> }<a name="line.243"></a> @@ -1029,155 +1029,163 @@ <span class="sourceLineNo">1021</span> /**<a name="line.1021"></a> <span class="sourceLineNo">1022</span> * May return 0 when server is not online.<a name="line.1022"></a> <span class="sourceLineNo">1023</span> */<a name="line.1023"></a> -<span class="sourceLineNo">1024</span> public int getServerVersion(final ServerName serverName) {<a name="line.1024"></a> +<span class="sourceLineNo">1024</span> public int getVersionNumber(ServerName serverName) {<a name="line.1024"></a> <span class="sourceLineNo">1025</span> ServerMetrics serverMetrics = onlineServers.get(serverName);<a name="line.1025"></a> <span class="sourceLineNo">1026</span> return serverMetrics != null ? serverMetrics.getVersionNumber() : 0;<a name="line.1026"></a> <span class="sourceLineNo">1027</span> }<a name="line.1027"></a> <span class="sourceLineNo">1028</span><a name="line.1028"></a> -<span class="sourceLineNo">1029</span> public int getInfoPort(ServerName serverName) {<a name="line.1029"></a> -<span class="sourceLineNo">1030</span> ServerMetrics serverMetrics = onlineServers.get(serverName);<a name="line.1030"></a> -<span class="sourceLineNo">1031</span> return serverMetrics != null ? serverMetrics.getInfoServerPort() : 0;<a name="line.1031"></a> -<span class="sourceLineNo">1032</span> }<a name="line.1032"></a> -<span class="sourceLineNo">1033</span><a name="line.1033"></a> -<span class="sourceLineNo">1034</span> /**<a name="line.1034"></a> -<span class="sourceLineNo">1035</span> * Persist last flushed sequence id of each region to HDFS<a name="line.1035"></a> -<span class="sourceLineNo">1036</span> * @throws IOException if persit to HDFS fails<a name="line.1036"></a> -<span class="sourceLineNo">1037</span> */<a name="line.1037"></a> -<span class="sourceLineNo">1038</span> private void persistRegionLastFlushedSequenceIds() throws IOException {<a name="line.1038"></a> -<span class="sourceLineNo">1039</span> if (isFlushSeqIdPersistInProgress) {<a name="line.1039"></a> -<span class="sourceLineNo">1040</span> return;<a name="line.1040"></a> -<span class="sourceLineNo">1041</span> }<a name="line.1041"></a> -<span class="sourceLineNo">1042</span> isFlushSeqIdPersistInProgress = true;<a name="line.1042"></a> -<span class="sourceLineNo">1043</span> try {<a name="line.1043"></a> -<span class="sourceLineNo">1044</span> Configuration conf = master.getConfiguration();<a name="line.1044"></a> -<span class="sourceLineNo">1045</span> Path rootDir = FSUtils.getRootDir(conf);<a name="line.1045"></a> -<span class="sourceLineNo">1046</span> Path lastFlushedSeqIdPath = new Path(rootDir, LAST_FLUSHED_SEQ_ID_FILE);<a name="line.1046"></a> -<span class="sourceLineNo">1047</span> FileSystem fs = FileSystem.get(conf);<a name="line.1047"></a> -<span class="sourceLineNo">1048</span> if (fs.exists(lastFlushedSeqIdPath)) {<a name="line.1048"></a> -<span class="sourceLineNo">1049</span> LOG.info("Rewriting .lastflushedseqids file at: "<a name="line.1049"></a> -<span class="sourceLineNo">1050</span> + lastFlushedSeqIdPath);<a name="line.1050"></a> -<span class="sourceLineNo">1051</span> if (!fs.delete(lastFlushedSeqIdPath, false)) {<a name="line.1051"></a> -<span class="sourceLineNo">1052</span> throw new IOException("Unable to remove existing "<a name="line.1052"></a> -<span class="sourceLineNo">1053</span> + lastFlushedSeqIdPath);<a name="line.1053"></a> -<span class="sourceLineNo">1054</span> }<a name="line.1054"></a> -<span class="sourceLineNo">1055</span> } else {<a name="line.1055"></a> -<span class="sourceLineNo">1056</span> LOG.info("Writing .lastflushedseqids file at: " + lastFlushedSeqIdPath);<a name="line.1056"></a> -<span class="sourceLineNo">1057</span> }<a name="line.1057"></a> -<span class="sourceLineNo">1058</span> FSDataOutputStream out = fs.create(lastFlushedSeqIdPath);<a name="line.1058"></a> -<span class="sourceLineNo">1059</span> FlushedSequenceId.Builder flushedSequenceIdBuilder =<a name="line.1059"></a> -<span class="sourceLineNo">1060</span> FlushedSequenceId.newBuilder();<a name="line.1060"></a> -<span class="sourceLineNo">1061</span> try {<a name="line.1061"></a> -<span class="sourceLineNo">1062</span> for (Entry<byte[], Long> entry : flushedSequenceIdByRegion.entrySet()) {<a name="line.1062"></a> -<span class="sourceLineNo">1063</span> FlushedRegionSequenceId.Builder flushedRegionSequenceIdBuilder =<a name="line.1063"></a> -<span class="sourceLineNo">1064</span> FlushedRegionSequenceId.newBuilder();<a name="line.1064"></a> -<span class="sourceLineNo">1065</span> flushedRegionSequenceIdBuilder.setRegionEncodedName(<a name="line.1065"></a> -<span class="sourceLineNo">1066</span> ByteString.copyFrom(entry.getKey()));<a name="line.1066"></a> -<span class="sourceLineNo">1067</span> flushedRegionSequenceIdBuilder.setSeqId(entry.getValue());<a name="line.1067"></a> -<span class="sourceLineNo">1068</span> ConcurrentNavigableMap<byte[], Long> storeSeqIds =<a name="line.1068"></a> -<span class="sourceLineNo">1069</span> storeFlushedSequenceIdsByRegion.get(entry.getKey());<a name="line.1069"></a> -<span class="sourceLineNo">1070</span> if (storeSeqIds != null) {<a name="line.1070"></a> -<span class="sourceLineNo">1071</span> for (Entry<byte[], Long> store : storeSeqIds.entrySet()) {<a name="line.1071"></a> -<span class="sourceLineNo">1072</span> FlushedStoreSequenceId.Builder flushedStoreSequenceIdBuilder =<a name="line.1072"></a> -<span class="sourceLineNo">1073</span> FlushedStoreSequenceId.newBuilder();<a name="line.1073"></a> -<span class="sourceLineNo">1074</span> flushedStoreSequenceIdBuilder.setFamily(ByteString.copyFrom(store.getKey()));<a name="line.1074"></a> -<span class="sourceLineNo">1075</span> flushedStoreSequenceIdBuilder.setSeqId(store.getValue());<a name="line.1075"></a> -<span class="sourceLineNo">1076</span> flushedRegionSequenceIdBuilder.addStores(flushedStoreSequenceIdBuilder);<a name="line.1076"></a> -<span class="sourceLineNo">1077</span> }<a name="line.1077"></a> -<span class="sourceLineNo">1078</span> }<a name="line.1078"></a> -<span class="sourceLineNo">1079</span> flushedSequenceIdBuilder.addRegionSequenceId(flushedRegionSequenceIdBuilder);<a name="line.1079"></a> -<span class="sourceLineNo">1080</span> }<a name="line.1080"></a> -<span class="sourceLineNo">1081</span> flushedSequenceIdBuilder.build().writeDelimitedTo(out);<a name="line.1081"></a> -<span class="sourceLineNo">1082</span> } finally {<a name="line.1082"></a> -<span class="sourceLineNo">1083</span> if (out != null) {<a name="line.1083"></a> -<span class="sourceLineNo">1084</span> out.close();<a name="line.1084"></a> -<span class="sourceLineNo">1085</span> }<a name="line.1085"></a> -<span class="sourceLineNo">1086</span> }<a name="line.1086"></a> -<span class="sourceLineNo">1087</span> } finally {<a name="line.1087"></a> -<span class="sourceLineNo">1088</span> isFlushSeqIdPersistInProgress = false;<a name="line.1088"></a> -<span class="sourceLineNo">1089</span> }<a name="line.1089"></a> -<span class="sourceLineNo">1090</span> }<a name="line.1090"></a> -<span class="sourceLineNo">1091</span><a name="line.1091"></a> -<span class="sourceLineNo">1092</span> /**<a name="line.1092"></a> -<span class="sourceLineNo">1093</span> * Load last flushed sequence id of each region from HDFS, if persisted<a name="line.1093"></a> -<span class="sourceLineNo">1094</span> */<a name="line.1094"></a> -<span class="sourceLineNo">1095</span> public void loadLastFlushedSequenceIds() throws IOException {<a name="line.1095"></a> -<span class="sourceLineNo">1096</span> if (!persistFlushedSequenceId) {<a name="line.1096"></a> -<span class="sourceLineNo">1097</span> return;<a name="line.1097"></a> -<span class="sourceLineNo">1098</span> }<a name="line.1098"></a> -<span class="sourceLineNo">1099</span> Configuration conf = master.getConfiguration();<a name="line.1099"></a> -<span class="sourceLineNo">1100</span> Path rootDir = FSUtils.getRootDir(conf);<a name="line.1100"></a> -<span class="sourceLineNo">1101</span> Path lastFlushedSeqIdPath = new Path(rootDir, LAST_FLUSHED_SEQ_ID_FILE);<a name="line.1101"></a> -<span class="sourceLineNo">1102</span> FileSystem fs = FileSystem.get(conf);<a name="line.1102"></a> -<span class="sourceLineNo">1103</span> if (!fs.exists(lastFlushedSeqIdPath)) {<a name="line.1103"></a> -<span class="sourceLineNo">1104</span> LOG.info("No .lastflushedseqids found at" + lastFlushedSeqIdPath<a name="line.1104"></a> -<span class="sourceLineNo">1105</span> + " will record last flushed sequence id"<a name="line.1105"></a> -<span class="sourceLineNo">1106</span> + " for regions by regionserver report all over again");<a name="line.1106"></a> -<span class="sourceLineNo">1107</span> return;<a name="line.1107"></a> -<span class="sourceLineNo">1108</span> } else {<a name="line.1108"></a> -<span class="sourceLineNo">1109</span> LOG.info("begin to load .lastflushedseqids at " + lastFlushedSeqIdPath);<a name="line.1109"></a> -<span class="sourceLineNo">1110</span> }<a name="line.1110"></a> -<span class="sourceLineNo">1111</span> FSDataInputStream in = fs.open(lastFlushedSeqIdPath);<a name="line.1111"></a> -<span class="sourceLineNo">1112</span> try {<a name="line.1112"></a> -<span class="sourceLineNo">1113</span> FlushedSequenceId flushedSequenceId =<a name="line.1113"></a> -<span class="sourceLineNo">1114</span> FlushedSequenceId.parseDelimitedFrom(in);<a name="line.1114"></a> -<span class="sourceLineNo">1115</span> for (FlushedRegionSequenceId flushedRegionSequenceId : flushedSequenceId<a name="line.1115"></a> -<span class="sourceLineNo">1116</span> .getRegionSequenceIdList()) {<a name="line.1116"></a> -<span class="sourceLineNo">1117</span> byte[] encodedRegionName = flushedRegionSequenceId<a name="line.1117"></a> -<span class="sourceLineNo">1118</span> .getRegionEncodedName().toByteArray();<a name="line.1118"></a> -<span class="sourceLineNo">1119</span> flushedSequenceIdByRegion<a name="line.1119"></a> -<span class="sourceLineNo">1120</span> .putIfAbsent(encodedRegionName, flushedRegionSequenceId.getSeqId());<a name="line.1120"></a> -<span class="sourceLineNo">1121</span> if (flushedRegionSequenceId.getStoresList() != null<a name="line.1121"></a> -<span class="sourceLineNo">1122</span> && flushedRegionSequenceId.getStoresList().size() != 0) {<a name="line.1122"></a> -<span class="sourceLineNo">1123</span> ConcurrentNavigableMap<byte[], Long> storeFlushedSequenceId =<a name="line.1123"></a> -<span class="sourceLineNo">1124</span> computeIfAbsent(storeFlushedSequenceIdsByRegion, encodedRegionName,<a name="line.1124"></a> -<span class="sourceLineNo">1125</span> () -> new ConcurrentSkipListMap<>(Bytes.BYTES_COMPARATOR));<a name="line.1125"></a> -<span class="sourceLineNo">1126</span> for (FlushedStoreSequenceId flushedStoreSequenceId : flushedRegionSequenceId<a name="line.1126"></a> -<span class="sourceLineNo">1127</span> .getStoresList()) {<a name="line.1127"></a> -<span class="sourceLineNo">1128</span> storeFlushedSequenceId<a name="line.1128"></a> -<span class="sourceLineNo">1129</span> .put(flushedStoreSequenceId.getFamily().toByteArray(),<a name="line.1129"></a> -<span class="sourceLineNo">1130</span> flushedStoreSequenceId.getSeqId());<a name="line.1130"></a> -<span class="sourceLineNo">1131</span> }<a name="line.1131"></a> -<span class="sourceLineNo">1132</span> }<a name="line.1132"></a> -<span class="sourceLineNo">1133</span> }<a name="line.1133"></a> -<span class="sourceLineNo">1134</span> } finally {<a name="line.1134"></a> -<span class="sourceLineNo">1135</span> in.close();<a name="line.1135"></a> -<span class="sourceLineNo">1136</span> }<a name="line.1136"></a> -<span class="sourceLineNo">1137</span> }<a name="line.1137"></a> -<span class="sourceLineNo">1138</span><a name="line.1138"></a> -<span class="sourceLineNo">1139</span> /**<a name="line.1139"></a> -<span class="sourceLineNo">1140</span> * Regions may have been removed between latest persist of FlushedSequenceIds<a name="line.1140"></a> -<span class="sourceLineNo">1141</span> * and master abort. So after loading FlushedSequenceIds from file, and after<a name="line.1141"></a> -<span class="sourceLineNo">1142</span> * meta loaded, we need to remove the deleted region according to RegionStates.<a name="line.1142"></a> -<span class="sourceLineNo">1143</span> */<a name="line.1143"></a> -<span class="sourceLineNo">1144</span> public void removeDeletedRegionFromLoadedFlushedSequenceIds() {<a name="line.1144"></a> -<span class="sourceLineNo">1145</span> RegionStates regionStates = master.getAssignmentManager().getRegionStates();<a name="line.1145"></a> -<span class="sourceLineNo">1146</span> Iterator<byte[]> it = flushedSequenceIdByRegion.keySet().iterator();<a name="line.1146"></a> -<span class="sourceLineNo">1147</span> while(it.hasNext()) {<a name="line.1147"></a> -<span class="sourceLineNo">1148</span> byte[] regionEncodedName = it.next();<a name="line.1148"></a> -<span class="sourceLineNo">1149</span> if (regionStates.getRegionState(Bytes.toStringBinary(regionEncodedName)) == null) {<a name="line.1149"></a> -<span class="sourceLineNo">1150</span> it.remove();<a name="line.1150"></a> -<span class="sourceLineNo">1151</span> storeFlushedSequenceIdsByRegion.remove(regionEncodedName);<a name="line.1151"></a> -<span class="sourceLineNo">1152</span> }<a name="line.1152"></a> -<span class="sourceLineNo">1153</span> }<a name="line.1153"></a> -<span class="sourceLineNo">1154</span> }<a name="line.1154"></a> -<span class="sourceLineNo">1155</span><a name="line.1155"></a> -<span class="sourceLineNo">1156</span> private class FlushedSequenceIdFlusher extends ScheduledChore {<a name="line.1156"></a> -<span class="sourceLineNo">1157</span><a name="line.1157"></a> -<span class="sourceLineNo">1158</span> public FlushedSequenceIdFlusher(String name, int p) {<a name="line.1158"></a> -<span class="sourceLineNo">1159</span> super(name, master, p, 60 * 1000); //delay one minute before first execute<a name="line.1159"></a> -<span class="sourceLineNo">1160</span> }<a name="line.1160"></a> -<span class="sourceLineNo">1161</span><a name="line.1161"></a> -<span class="sourceLineNo">1162</span> @Override<a name="line.1162"></a> -<span class="sourceLineNo">1163</span> protected void chore() {<a name="line.1163"></a> -<span class="sourceLineNo">1164</span> try {<a name="line.1164"></a> -<span class="sourceLineNo">1165</span> persistRegionLastFlushedSequenceIds();<a name="line.1165"></a> -<span class="sourceLineNo">1166</span> } catch (IOException e) {<a name="line.1166"></a> -<span class="sourceLineNo">1167</span> LOG.debug("Failed to persist last flushed sequence id of regions"<a name="line.1167"></a> -<span class="sourceLineNo">1168</span> + " to file system", e);<a name="line.1168"></a> -<span class="sourceLineNo">1169</span> }<a name="line.1169"></a> -<span class="sourceLineNo">1170</span> }<a name="line.1170"></a> -<span class="sourceLineNo">1171</span> }<a name="line.1171"></a> -<span class="sourceLineNo">1172</span>}<a name="line.1172"></a> +<span class="sourceLineNo">1029</span> /**<a name="line.1029"></a> +<span class="sourceLineNo">1030</span> * May return "0.0.0" when server is not online<a name="line.1030"></a> +<span class="sourceLineNo">1031</span> */<a name="line.1031"></a> +<span class="sourceLineNo">1032</span> public String getVersion(ServerName serverName) {<a name="line.1032"></a> +<span class="sourceLineNo">1033</span> ServerMetrics serverMetrics = onlineServers.get(serverName);<a name="line.1033"></a> +<span class="sourceLineNo">1034</span> return serverMetrics != null ? serverMetrics.getVersion() : "0.0.0";<a name="line.1034"></a> +<span class="sourceLineNo">1035</span> }<a name="line.1035"></a> +<span class="sourceLineNo">1036</span><a name="line.1036"></a> +<span class="sourceLineNo">1037</span> public int getInfoPort(ServerName serverName) {<a name="line.1037"></a> +<span class="sourceLineNo">1038</span> ServerMetrics serverMetrics = onlineServers.get(serverName);<a name="line.1038"></a> +<span class="sourceLineNo">1039</span> return serverMetrics != null ? serverMetrics.getInfoServerPort() : 0;<a name="line.1039"></a> +<span class="sourceLineNo">1040</span> }<a name="line.1040"></a> +<span class="sourceLineNo">1041</span><a name="line.1041"></a> +<span class="sourceLineNo">1042</span> /**<a name="line.1042"></a> +<span class="sourceLineNo">1043</span> * Persist last flushed sequence id of each region to HDFS<a name="line.1043"></a> +<span class="sourceLineNo">1044</span> * @throws IOException if persit to HDFS fails<a name="line.1044"></a> +<span class="sourceLineNo">1045</span> */<a name="line.1045"></a> +<span class="sourceLineNo">1046</span> private void persistRegionLastFlushedSequenceIds() throws IOException {<a name="line.1046"></a> +<span class="sourceLineNo">1047</span> if (isFlushSeqIdPersistInProgress) {<a name="line.1047"></a> +<span class="sourceLineNo">1048</span> return;<a name="line.1048"></a> +<span class="sourceLineNo">1049</span> }<a name="line.1049"></a> +<span class="sourceLineNo">1050</span> isFlushSeqIdPersistInProgress = true;<a name="line.1050"></a> +<span class="sourceLineNo">1051</span> try {<a name="line.1051"></a> +<span class="sourceLineNo">1052</span> Configuration conf = master.getConfiguration();<a name="line.1052"></a> +<span class="sourceLineNo">1053</span> Path rootDir = FSUtils.getRootDir(conf);<a name="line.1053"></a> +<span class="sourceLineNo">1054</span> Path lastFlushedSeqIdPath = new Path(rootDir, LAST_FLUSHED_SEQ_ID_FILE);<a name="line.1054"></a> +<span class="sourceLineNo">1055</span> FileSystem fs = FileSystem.get(conf);<a name="line.1055"></a> +<span class="sourceLineNo">1056</span> if (fs.exists(lastFlushedSeqIdPath)) {<a name="line.1056"></a> +<span class="sourceLineNo">1057</span> LOG.info("Rewriting .lastflushedseqids file at: "<a name="line.1057"></a> +<span class="sourceLineNo">1058</span> + lastFlushedSeqIdPath);<a name="line.1058"></a> +<span class="sourceLineNo">1059</span> if (!fs.delete(lastFlushedSeqIdPath, false)) {<a name="line.1059"></a> +<span class="sourceLineNo">1060</span> throw new IOException("Unable to remove existing "<a name="line.1060"></a> +<span class="sourceLineNo">1061</span> + lastFlushedSeqIdPath);<a name="line.1061"></a> +<span class="sourceLineNo">1062</span> }<a name="line.1062"></a> +<span class="sourceLineNo">1063</span> } else {<a name="line.1063"></a> +<span class="sourceLineNo">1064</span> LOG.info("Writing .lastflushedseqids file at: " + lastFlushedSeqIdPath);<a name="line.1064"></a> +<span class="sourceLineNo">1065</span> }<a name="line.1065"></a> +<span class="sourceLineNo">1066</span> FSDataOutputStream out = fs.create(lastFlushedSeqIdPath);<a name="line.1066"></a> +<span class="sourceLineNo">1067</span> FlushedSequenceId.Builder flushedSequenceIdBuilder =<a name="line.1067"></a> +<span class="sourceLineNo">1068</span> FlushedSequenceId.newBuilder();<a name="line.1068"></a> +<span class="sourceLineNo">1069</span> try {<a name="line.1069"></a> +<span class="sourceLineNo">1070</span> for (Entry<byte[], Long> entry : flushedSequenceIdByRegion.entrySet()) {<a name="line.1070"></a> +<span class="sourceLineNo">1071</span> FlushedRegionSequenceId.Builder flushedRegionSequenceIdBuilder =<a name="line.1071"></a> +<span class="sourceLineNo">1072</span> FlushedRegionSequenceId.newBuilder();<a name="line.1072"></a> +<span class="sourceLineNo">1073</span> flushedRegionSequenceIdBuilder.setRegionEncodedName(<a name="line.1073"></a> +<span class="sourceLineNo">1074</span> ByteString.copyFrom(entry.getKey()));<a name="line.1074"></a> +<span class="sourceLineNo">1075</span> flushedRegionSequenceIdBuilder.setSeqId(entry.getValue());<a name="line.1075"></a> +<span class="sourceLineNo">1076</span> ConcurrentNavigableMap<byte[], Long> storeSeqIds =<a name="line.1076"></a> +<span class="sourceLineNo">1077</span> storeFlushedSequenceIdsByRegion.get(entry.getKey());<a name="line.1077"></a> +<span class="sourceLineNo">1078</span> if (storeSeqIds != null) {<a name="line.1078"></a> +<span class="sourceLineNo">1079</span> for (Entry<byte[], Long> store : storeSeqIds.entrySet()) {<a name="line.1079"></a> +<span class="sourceLineNo">1080</span> FlushedStoreSequenceId.Builder flushedStoreSequenceIdBuilder =<a name="line.1080"></a> +<span class="sourceLineNo">1081</span> FlushedStoreSequenceId.newBuilder();<a name="line.1081"></a> +<span class="sourceLineNo">1082</span> flushedStoreSequenceIdBuilder.setFamily(ByteString.copyFrom(store.getKey()));<a name="line.1082"></a> +<span class="sourceLineNo">1083</span> flushedStoreSequenceIdBuilder.setSeqId(store.getValue());<a name="line.1083"></a> +<span class="sourceLineNo">1084</span> flushedRegionSequenceIdBuilder.addStores(flushedStoreSequenceIdBuilder);<a name="line.1084"></a> +<span class="sourceLineNo">1085</span> }<a name="line.1085"></a> +<span class="sourceLineNo">1086</span> }<a name="line.1086"></a> +<span class="sourceLineNo">1087</span> flushedSequenceIdBuilder.addRegionSequenceId(flushedRegionSequenceIdBuilder);<a name="line.1087"></a> +<span class="sourceLineNo">1088</span> }<a name="line.1088"></a> +<span class="sourceLineNo">1089</span> flushedSequenceIdBuilder.build().writeDelimitedTo(out);<a name="line.1089"></a> +<span class="sourceLineNo">1090</span> } finally {<a name="line.1090"></a> +<span class="sourceLineNo">1091</span> if (out != null) {<a name="line.1091"></a> +<span class="sourceLineNo">1092</span> out.close();<a name="line.1092"></a> +<span class="sourceLineNo">1093</span> }<a name="line.1093"></a> +<span class="sourceLineNo">1094</span> }<a name="line.1094"></a> +<span class="sourceLineNo">1095</span> } finally {<a name="line.1095"></a> +<span class="sourceLineNo">1096</span> isFlushSeqIdPersistInProgress = false;<a name="line.1096"></a> +<span class="sourceLineNo">1097</span> }<a name="line.1097"></a> +<span class="sourceLineNo">1098</span> }<a name="line.1098"></a> +<span class="sourceLineNo">1099</span><a name="line.1099"></a> +<span class="sourceLineNo">1100</span> /**<a name="line.1100"></a> +<span class="sourceLineNo">1101</span> * Load last flushed sequence id of each region from HDFS, if persisted<a name="line.1101"></a> +<span class="sourceLineNo">1102</span> */<a name="line.1102"></a> +<span class="sourceLineNo">1103</span> public void loadLastFlushedSequenceIds() throws IOException {<a name="line.1103"></a> +<span class="sourceLineNo">1104</span> if (!persistFlushedSequenceId) {<a name="line.1104"></a> +<span class="sourceLineNo">1105</span> return;<a name="line.1105"></a> +<span class="sourceLineNo">1106</span> }<a name="line.1106"></a> +<span class="sourceLineNo">1107</span> Configuration conf = master.getConfiguration();<a name="line.1107"></a> +<span class="sourceLineNo">1108</span> Path rootDir = FSUtils.getRootDir(conf);<a name="line.1108"></a> +<span class="sourceLineNo">1109</span> Path lastFlushedSeqIdPath = new Path(rootDir, LAST_FLUSHED_SEQ_ID_FILE);<a name="line.1109"></a> +<span class="sourceLineNo">1110</span> FileSystem fs = FileSystem.get(conf);<a name="line.1110"></a> +<span class="sourceLineNo">1111</span> if (!fs.exists(lastFlushedSeqIdPath)) {<a name="line.1111"></a> +<span class="sourceLineNo">1112</span> LOG.info("No .lastflushedseqids found at" + lastFlushedSeqIdPath<a name="line.1112"></a> +<span class="sourceLineNo">1113</span> + " will record last flushed sequence id"<a name="line.1113"></a> +<span class="sourceLineNo">1114</span> + " for regions by regionserver report all over again");<a name="line.1114"></a> +<span class="sourceLineNo">1115</span> return;<a name="line.1115"></a> +<span class="sourceLineNo">1116</span> } else {<a name="line.1116"></a> +<span class="sourceLineNo">1117</span> LOG.info("begin to load .lastflushedseqids at " + lastFlushedSeqIdPath);<a name="line.1117"></a> +<span class="sourceLineNo">1118</span> }<a name="line.1118"></a> +<span class="sourceLineNo">1119</span> FSDataInputStream in = fs.open(lastFlushedSeqIdPath);<a name="line.1119"></a> +<span class="sourceLineNo">1120</span> try {<a name="line.1120"></a> +<span class="sourceLineNo">1121</span> FlushedSequenceId flushedSequenceId =<a name="line.1121"></a> +<span class="sourceLineNo">1122</span> FlushedSequenceId.parseDelimitedFrom(in);<a name="line.1122"></a> +<span class="sourceLineNo">1123</span> for (FlushedRegionSequenceId flushedRegionSequenceId : flushedSequenceId<a name="line.1123"></a> +<span class="sourceLineNo">1124</span> .getRegionSequenceIdList()) {<a name="line.1124"></a> +<span class="sourceLineNo">1125</span> byte[] encodedRegionName = flushedRegionSequenceId<a name="line.1125"></a> +<span class="sourceLineNo">1126</span> .getRegionEncodedName().toByteArray();<a name="line.1126"></a> +<span class="sourceLineNo">1127</span> flushedSequenceIdByRegion<a name="line.1127"></a> +<span class="sourceLineNo">1128</span> .putIfAbsent(encodedRegionName, flushedRegionSequenceId.getSeqId());<a name="line.1128"></a> +<span class="sourceLineNo">1129</span> if (flushedRegionSequenceId.getStoresList() != null<a name="line.1129"></a> +<span class="sourceLineNo">1130</span> && flushedRegionSequenceId.getStoresList().size() != 0) {<a name="line.1130"></a> +<span class="sourceLineNo">1131</span> ConcurrentNavigableMap<byte[], Long> storeFlushedSequenceId =<a name="line.1131"></a> +<span class="sourceLineNo">1132</span> computeIfAbsent(storeFlushedSequenceIdsByRegion, encodedRegionName,<a name="line.1132"></a> +<span class="sourceLineNo">1133</span> () -> new ConcurrentSkipListMap<>(Bytes.BYTES_COMPARATOR));<a name="line.1133"></a> +<span class="sourceLineNo">1134</span> for (FlushedStoreSequenceId flushedStoreSequenceId : flushedRegionSequenceId<a name="line.1134"></a> +<span class="sourceLineNo">1135</span> .getStoresList()) {<a name="line.1135"></a> +<span class="sourceLineNo">1136</span> storeFlushedSequenceId<a name="line.1136"></a> +<span class="sourceLineNo">1137</span> .put(flushedStoreSequenceId.getFamily().toByteArray(),<a name="line.1137"></a> +<span class="sourceLineNo">1138</span> flushedStoreSequenceId.getSeqId());<a name="line.1138"></a> +<span class="sourceLineNo">1139</span> }<a name="line.1139"></a> +<span class="sourceLineNo">1140</span> }<a name="line.1140"></a> +<span class="sourceLineNo">1141</span> }<a name="line.1141"></a> +<span class="sourceLineNo">1142</span> } finally {<a name="line.1142"></a> +<span class="sourceLineNo">1143</span> in.close();<a name="line.1143"></a> +<span class="sourceLineNo">1144</span> }<a name="line.1144"></a> +<span class="sourceLineNo">1145</span> }<a name="line.1145"></a> +<span class="sourceLineNo">1146</span><a name="line.1146"></a> +<span class="sourceLineNo">1147</span> /**<a name="line.1147"></a> +<span class="sourceLineNo">1148</span> * Regions may have been removed between latest persist of FlushedSequenceIds<a name="line.1148"></a> +<span class="sourceLineNo">1149</span> * and master abort. So after loading FlushedSequenceIds from file, and after<a name="line.1149"></a> +<span class="sourceLineNo">1150</span> * meta loaded, we need to remove the deleted region according to RegionStates.<a name="line.1150"></a> +<span class="sourceLineNo">1151</span> */<a name="line.1151"></a> +<span class="sourceLineNo">1152</span> public void removeDeletedRegionFromLoadedFlushedSequenceIds() {<a name="line.1152"></a> +<span class="sourceLineNo">1153</span> RegionStates regionStates = master.getAssignmentManager().getRegionStates();<a name="line.1153"></a> +<span class="sourceLineNo">1154</span> Iterator<byte[]> it = flushedSequenceIdByRegion.keySet().iterator();<a name="line.1154"></a> +<span class="sourceLineNo">1155</span> while(it.hasNext()) {<a name="line.1155"></a> +<span class="sourceLineNo">1156</span> byte[] regionEncodedName = it.next();<a name="line.1156"></a> +<span class="sourceLineNo">1157</span> if (regionStates.getRegionState(Bytes.toStringBinary(regionEncodedName)) == null) {<a name="line.1157"></a> +<span class="sourceLineNo">1158</span> it.remove();<a name="line.1158"></a> +<span class="sourceLineNo">1159</span> storeFlushedSequenceIdsByRegion.remove(regionEncodedName);<a name="line.1159"></a> +<span class="sourceLineNo">1160</span> }<a name="line.1160"></a> +<span class="sourceLineNo">1161</span> }<a name="line.1161"></a> +<span class="sourceLineNo">1162</span> }<a name="line.1162"></a> +<span class="sourceLineNo">1163</span><a name="line.1163"></a> +<span class="sourceLineNo">1164</span> private class FlushedSequenceIdFlusher extends ScheduledChore {<a name="line.1164"></a> +<span class="sourceLineNo">1165</span><a name="line.1165"></a> +<span class="sourceLineNo">1166</span> public FlushedSequenceIdFlusher(String name, int p) {<a name="line.1166"></a> +<span class="sourceLineNo">1167</span> super(name, master, p, 60 * 1000); //delay one minute before first execute<a name="line.1167"></a> +<span class="sourceLineNo">1168</span> }<a name="line.1168"></a> +<span class="sourceLineNo">1169</span><a name="line.1169"></a> +<span class="sourceLineNo">1170</span> @Override<a name="line.1170"></a> +<span class="sourceLineNo">1171</span> protected void chore() {<a name="line.1171"></a> +<span class="sourceLineNo">1172</span> try {<a name="line.1172"></a> +<span class="sourceLineNo">1173</span> persistRegionLastFlushedSequenceIds();<a name="line.1173"></a> +<span class="sourceLineNo">1174</span> } catch (IOException e) {<a name="line.1174"></a> +<span class="sourceLineNo">1175</span> LOG.debug("Failed to persist last flushed sequence id of regions"<a name="line.1175"></a> +<span class="sourceLineNo">1176</span> + " to file system", e);<a name="line.1176"></a> +<span class="sourceLineNo">1177</span> }<a name="line.1177"></a> +<span class="sourceLineNo">1178</span> }<a name="line.1178"></a> +<span class="sourceLineNo">1179</span> }<a name="line.1179"></a> +<span class="sourceLineNo">1180</span>}<a name="line.1180"></a>
