blob: e4a5e86c4c2fed053c80758f50b2457bcd8eaaf2 [file] [log] [blame]
<!DOCTYPE HTML>
<html lang="en">
<head>
<!-- Generated by javadoc (17) -->
<title>Source code</title>
<meta name="viewport" content="width=device-width, initial-scale=1">
<meta name="description" content="source: package: org.apache.hadoop.hbase.regionserver.wal, class: AbstractProtobufWALReader">
<meta name="generator" content="javadoc/SourceToHTMLConverter">
<link rel="stylesheet" type="text/css" href="../../../../../../../stylesheet.css" title="Style">
</head>
<body class="source-page">
<main role="main">
<div class="source-container">
<pre><span class="source-line-no">001</span><span id="line-1">/*</span>
<span class="source-line-no">002</span><span id="line-2"> * Licensed to the Apache Software Foundation (ASF) under one</span>
<span class="source-line-no">003</span><span id="line-3"> * or more contributor license agreements. See the NOTICE file</span>
<span class="source-line-no">004</span><span id="line-4"> * distributed with this work for additional information</span>
<span class="source-line-no">005</span><span id="line-5"> * regarding copyright ownership. The ASF licenses this file</span>
<span class="source-line-no">006</span><span id="line-6"> * to you under the Apache License, Version 2.0 (the</span>
<span class="source-line-no">007</span><span id="line-7"> * "License"); you may not use this file except in compliance</span>
<span class="source-line-no">008</span><span id="line-8"> * with the License. You may obtain a copy of the License at</span>
<span class="source-line-no">009</span><span id="line-9"> *</span>
<span class="source-line-no">010</span><span id="line-10"> * http://www.apache.org/licenses/LICENSE-2.0</span>
<span class="source-line-no">011</span><span id="line-11"> *</span>
<span class="source-line-no">012</span><span id="line-12"> * Unless required by applicable law or agreed to in writing, software</span>
<span class="source-line-no">013</span><span id="line-13"> * distributed under the License is distributed on an "AS IS" BASIS,</span>
<span class="source-line-no">014</span><span id="line-14"> * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.</span>
<span class="source-line-no">015</span><span id="line-15"> * See the License for the specific language governing permissions and</span>
<span class="source-line-no">016</span><span id="line-16"> * limitations under the License.</span>
<span class="source-line-no">017</span><span id="line-17"> */</span>
<span class="source-line-no">018</span><span id="line-18">package org.apache.hadoop.hbase.regionserver.wal;</span>
<span class="source-line-no">019</span><span id="line-19"></span>
<span class="source-line-no">020</span><span id="line-20">import java.io.Closeable;</span>
<span class="source-line-no">021</span><span id="line-21">import java.io.EOFException;</span>
<span class="source-line-no">022</span><span id="line-22">import java.io.FileNotFoundException;</span>
<span class="source-line-no">023</span><span id="line-23">import java.io.IOException;</span>
<span class="source-line-no">024</span><span id="line-24">import java.io.InputStream;</span>
<span class="source-line-no">025</span><span id="line-25">import java.nio.ByteBuffer;</span>
<span class="source-line-no">026</span><span id="line-26">import java.security.Key;</span>
<span class="source-line-no">027</span><span id="line-27">import java.security.KeyException;</span>
<span class="source-line-no">028</span><span id="line-28">import java.util.Arrays;</span>
<span class="source-line-no">029</span><span id="line-29">import java.util.List;</span>
<span class="source-line-no">030</span><span id="line-30">import org.apache.commons.io.IOUtils;</span>
<span class="source-line-no">031</span><span id="line-31">import org.apache.hadoop.conf.Configuration;</span>
<span class="source-line-no">032</span><span id="line-32">import org.apache.hadoop.fs.FSDataInputStream;</span>
<span class="source-line-no">033</span><span id="line-33">import org.apache.hadoop.fs.FileStatus;</span>
<span class="source-line-no">034</span><span id="line-34">import org.apache.hadoop.fs.FileSystem;</span>
<span class="source-line-no">035</span><span id="line-35">import org.apache.hadoop.fs.Path;</span>
<span class="source-line-no">036</span><span id="line-36">import org.apache.hadoop.hbase.HConstants;</span>
<span class="source-line-no">037</span><span id="line-37">import org.apache.hadoop.hbase.codec.Codec;</span>
<span class="source-line-no">038</span><span id="line-38">import org.apache.hadoop.hbase.io.compress.Compression;</span>
<span class="source-line-no">039</span><span id="line-39">import org.apache.hadoop.hbase.io.crypto.Cipher;</span>
<span class="source-line-no">040</span><span id="line-40">import org.apache.hadoop.hbase.io.crypto.Decryptor;</span>
<span class="source-line-no">041</span><span id="line-41">import org.apache.hadoop.hbase.io.crypto.Encryption;</span>
<span class="source-line-no">042</span><span id="line-42">import org.apache.hadoop.hbase.io.util.LRUDictionary;</span>
<span class="source-line-no">043</span><span id="line-43">import org.apache.hadoop.hbase.security.EncryptionUtil;</span>
<span class="source-line-no">044</span><span id="line-44">import org.apache.hadoop.hbase.security.User;</span>
<span class="source-line-no">045</span><span id="line-45">import org.apache.hadoop.hbase.util.Bytes;</span>
<span class="source-line-no">046</span><span id="line-46">import org.apache.hadoop.hbase.util.CommonFSUtils;</span>
<span class="source-line-no">047</span><span id="line-47">import org.apache.hadoop.hbase.util.EncryptionTest;</span>
<span class="source-line-no">048</span><span id="line-48">import org.apache.hadoop.hbase.util.Pair;</span>
<span class="source-line-no">049</span><span id="line-49">import org.apache.hadoop.hbase.wal.AbstractFSWALProvider;</span>
<span class="source-line-no">050</span><span id="line-50">import org.apache.hadoop.ipc.RemoteException;</span>
<span class="source-line-no">051</span><span id="line-51">import org.apache.yetus.audience.InterfaceAudience;</span>
<span class="source-line-no">052</span><span id="line-52">import org.slf4j.Logger;</span>
<span class="source-line-no">053</span><span id="line-53">import org.slf4j.LoggerFactory;</span>
<span class="source-line-no">054</span><span id="line-54"></span>
<span class="source-line-no">055</span><span id="line-55">import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableList;</span>
<span class="source-line-no">056</span><span id="line-56">import org.apache.hbase.thirdparty.com.google.common.io.Closeables;</span>
<span class="source-line-no">057</span><span id="line-57">import org.apache.hbase.thirdparty.com.google.protobuf.InvalidProtocolBufferException;</span>
<span class="source-line-no">058</span><span id="line-58"></span>
<span class="source-line-no">059</span><span id="line-59">import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;</span>
<span class="source-line-no">060</span><span id="line-60">import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos;</span>
<span class="source-line-no">061</span><span id="line-61">import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos.WALTrailer;</span>
<span class="source-line-no">062</span><span id="line-62"></span>
<span class="source-line-no">063</span><span id="line-63">/**</span>
<span class="source-line-no">064</span><span id="line-64"> * Base class for reading protobuf based wal reader</span>
<span class="source-line-no">065</span><span id="line-65"> */</span>
<span class="source-line-no">066</span><span id="line-66">@InterfaceAudience.Private</span>
<span class="source-line-no">067</span><span id="line-67">public abstract class AbstractProtobufWALReader</span>
<span class="source-line-no">068</span><span id="line-68"> implements AbstractFSWALProvider.Initializer, Closeable {</span>
<span class="source-line-no">069</span><span id="line-69"></span>
<span class="source-line-no">070</span><span id="line-70"> private static final Logger LOG = LoggerFactory.getLogger(AbstractProtobufWALReader.class);</span>
<span class="source-line-no">071</span><span id="line-71"></span>
<span class="source-line-no">072</span><span id="line-72"> // public for WALFactory until we move everything to o.a.h.h.wal</span>
<span class="source-line-no">073</span><span id="line-73"> public static final byte[] PB_WAL_MAGIC = Bytes.toBytes("PWAL");</span>
<span class="source-line-no">074</span><span id="line-74"></span>
<span class="source-line-no">075</span><span id="line-75"> // public for TestWALSplit</span>
<span class="source-line-no">076</span><span id="line-76"> public static final byte[] PB_WAL_COMPLETE_MAGIC = Bytes.toBytes("LAWP");</span>
<span class="source-line-no">077</span><span id="line-77"></span>
<span class="source-line-no">078</span><span id="line-78"> /**</span>
<span class="source-line-no">079</span><span id="line-79"> * Configuration name of WAL Trailer's warning size. If a waltrailer's size is greater than the</span>
<span class="source-line-no">080</span><span id="line-80"> * configured size, providers should log a warning. e.g. this is used with Protobuf reader/writer.</span>
<span class="source-line-no">081</span><span id="line-81"> */</span>
<span class="source-line-no">082</span><span id="line-82"> static final String WAL_TRAILER_WARN_SIZE = "hbase.regionserver.waltrailer.warn.size";</span>
<span class="source-line-no">083</span><span id="line-83"> static final int DEFAULT_WAL_TRAILER_WARN_SIZE = 1024 * 1024; // 1MB</span>
<span class="source-line-no">084</span><span id="line-84"></span>
<span class="source-line-no">085</span><span id="line-85"> private static final List&lt;String&gt; WRITER_CLS_NAMES = ImmutableList.of(</span>
<span class="source-line-no">086</span><span id="line-86"> ProtobufLogWriter.class.getSimpleName(), AsyncProtobufLogWriter.class.getSimpleName(),</span>
<span class="source-line-no">087</span><span id="line-87"> "SecureProtobufLogWriter", "SecureAsyncProtobufLogWriter");</span>
<span class="source-line-no">088</span><span id="line-88"></span>
<span class="source-line-no">089</span><span id="line-89"> protected Configuration conf;</span>
<span class="source-line-no">090</span><span id="line-90"></span>
<span class="source-line-no">091</span><span id="line-91"> protected FileSystem fs;</span>
<span class="source-line-no">092</span><span id="line-92"></span>
<span class="source-line-no">093</span><span id="line-93"> protected Path path;</span>
<span class="source-line-no">094</span><span id="line-94"></span>
<span class="source-line-no">095</span><span id="line-95"> protected long fileLength;</span>
<span class="source-line-no">096</span><span id="line-96"></span>
<span class="source-line-no">097</span><span id="line-97"> protected FSDataInputStream inputStream;</span>
<span class="source-line-no">098</span><span id="line-98"></span>
<span class="source-line-no">099</span><span id="line-99"> protected CompressionContext compressionCtx;</span>
<span class="source-line-no">100</span><span id="line-100"> protected boolean hasCompression = false;</span>
<span class="source-line-no">101</span><span id="line-101"> protected boolean hasTagCompression = false;</span>
<span class="source-line-no">102</span><span id="line-102"> protected boolean hasValueCompression = false;</span>
<span class="source-line-no">103</span><span id="line-103"> protected Compression.Algorithm valueCompressionType;</span>
<span class="source-line-no">104</span><span id="line-104"></span>
<span class="source-line-no">105</span><span id="line-105"> protected Codec.Decoder cellDecoder;</span>
<span class="source-line-no">106</span><span id="line-106"> protected WALCellCodec.ByteStringUncompressor byteStringUncompressor;</span>
<span class="source-line-no">107</span><span id="line-107"></span>
<span class="source-line-no">108</span><span id="line-108"> protected long walEditsStopOffset;</span>
<span class="source-line-no">109</span><span id="line-109"> protected boolean trailerPresent;</span>
<span class="source-line-no">110</span><span id="line-110"> protected WALTrailer trailer;</span>
<span class="source-line-no">111</span><span id="line-111"> // maximum size of the wal Trailer in bytes. If a user writes/reads a trailer with size larger</span>
<span class="source-line-no">112</span><span id="line-112"> // than this size, it is written/read respectively, with a WARN message in the log.</span>
<span class="source-line-no">113</span><span id="line-113"> protected int trailerWarnSize;</span>
<span class="source-line-no">114</span><span id="line-114"></span>
<span class="source-line-no">115</span><span id="line-115"> // cell codec classname</span>
<span class="source-line-no">116</span><span id="line-116"> protected String codecClsName;</span>
<span class="source-line-no">117</span><span id="line-117"></span>
<span class="source-line-no">118</span><span id="line-118"> protected Decryptor decryptor;</span>
<span class="source-line-no">119</span><span id="line-119"></span>
<span class="source-line-no">120</span><span id="line-120"> /**</span>
<span class="source-line-no">121</span><span id="line-121"> * Get or create the input stream used by cell decoder.</span>
<span class="source-line-no">122</span><span id="line-122"> * &lt;p/&gt;</span>
<span class="source-line-no">123</span><span id="line-123"> * For implementing replication, we may need to limit the bytes we can read, so here we provide a</span>
<span class="source-line-no">124</span><span id="line-124"> * method so subclasses can wrap the original input stream.</span>
<span class="source-line-no">125</span><span id="line-125"> */</span>
<span class="source-line-no">126</span><span id="line-126"> protected abstract InputStream getCellCodecInputStream(FSDataInputStream stream);</span>
<span class="source-line-no">127</span><span id="line-127"></span>
<span class="source-line-no">128</span><span id="line-128"> /**</span>
<span class="source-line-no">129</span><span id="line-129"> * Skip to the given position.</span>
<span class="source-line-no">130</span><span id="line-130"> */</span>
<span class="source-line-no">131</span><span id="line-131"> protected abstract void skipTo(long position) throws IOException;</span>
<span class="source-line-no">132</span><span id="line-132"></span>
<span class="source-line-no">133</span><span id="line-133"> @Override</span>
<span class="source-line-no">134</span><span id="line-134"> public void init(FileSystem fs, Path path, Configuration conf, long startPosition)</span>
<span class="source-line-no">135</span><span id="line-135"> throws IOException {</span>
<span class="source-line-no">136</span><span id="line-136"> this.conf = conf;</span>
<span class="source-line-no">137</span><span id="line-137"> this.path = path;</span>
<span class="source-line-no">138</span><span id="line-138"> this.fs = fs;</span>
<span class="source-line-no">139</span><span id="line-139"> this.trailerWarnSize = conf.getInt(WAL_TRAILER_WARN_SIZE, DEFAULT_WAL_TRAILER_WARN_SIZE);</span>
<span class="source-line-no">140</span><span id="line-140"></span>
<span class="source-line-no">141</span><span id="line-141"> Pair&lt;FSDataInputStream, FileStatus&gt; pair = open();</span>
<span class="source-line-no">142</span><span id="line-142"> FSDataInputStream stream = pair.getFirst();</span>
<span class="source-line-no">143</span><span id="line-143"> FileStatus stat = pair.getSecond();</span>
<span class="source-line-no">144</span><span id="line-144"> boolean initSucceeded = false;</span>
<span class="source-line-no">145</span><span id="line-145"> try {</span>
<span class="source-line-no">146</span><span id="line-146"> // read the header</span>
<span class="source-line-no">147</span><span id="line-147"> WALProtos.WALHeader header = readHeader(stream);</span>
<span class="source-line-no">148</span><span id="line-148"> // initialize metadata and fields</span>
<span class="source-line-no">149</span><span id="line-149"> initDecryptor(header);</span>
<span class="source-line-no">150</span><span id="line-150"> initCompression(header);</span>
<span class="source-line-no">151</span><span id="line-151"> initWALCellCodec(header, getCellCodecInputStream(stream));</span>
<span class="source-line-no">152</span><span id="line-152"></span>
<span class="source-line-no">153</span><span id="line-153"> // read trailer if available</span>
<span class="source-line-no">154</span><span id="line-154"> readTrailer(stream, stat);</span>
<span class="source-line-no">155</span><span id="line-155"></span>
<span class="source-line-no">156</span><span id="line-156"> // this is intentional as we do not want the above methods to use the inputStream field. For</span>
<span class="source-line-no">157</span><span id="line-157"> // implementation tailing reader, we need to wrap the input stream when creating cell decoder,</span>
<span class="source-line-no">158</span><span id="line-158"> // so we need to make sure in the above methods we do not accidentally use the stored</span>
<span class="source-line-no">159</span><span id="line-159"> // inputStream directly and cause trouble. If a method needs to use an input stream, we just</span>
<span class="source-line-no">160</span><span id="line-160"> // pass the input stream in, like readHeader and readTrailer.</span>
<span class="source-line-no">161</span><span id="line-161"> this.inputStream = stream;</span>
<span class="source-line-no">162</span><span id="line-162"></span>
<span class="source-line-no">163</span><span id="line-163"> // seek to the given position if it is not -1</span>
<span class="source-line-no">164</span><span id="line-164"> if (startPosition &gt;= 0 &amp;&amp; startPosition != inputStream.getPos()) {</span>
<span class="source-line-no">165</span><span id="line-165"> if (compressionCtx != null) {</span>
<span class="source-line-no">166</span><span id="line-166"> // skip to the position, as we need to construct the compression dictionary</span>
<span class="source-line-no">167</span><span id="line-167"> skipTo(startPosition);</span>
<span class="source-line-no">168</span><span id="line-168"> } else {</span>
<span class="source-line-no">169</span><span id="line-169"> // just seek to the position</span>
<span class="source-line-no">170</span><span id="line-170"> stream.seek(startPosition);</span>
<span class="source-line-no">171</span><span id="line-171"> }</span>
<span class="source-line-no">172</span><span id="line-172"> }</span>
<span class="source-line-no">173</span><span id="line-173"> initSucceeded = true;</span>
<span class="source-line-no">174</span><span id="line-174"> } finally {</span>
<span class="source-line-no">175</span><span id="line-175"> if (!initSucceeded) {</span>
<span class="source-line-no">176</span><span id="line-176"> Closeables.close(stream, initSucceeded);</span>
<span class="source-line-no">177</span><span id="line-177"> inputStream = null;</span>
<span class="source-line-no">178</span><span id="line-178"> }</span>
<span class="source-line-no">179</span><span id="line-179"> }</span>
<span class="source-line-no">180</span><span id="line-180"> }</span>
<span class="source-line-no">181</span><span id="line-181"></span>
<span class="source-line-no">182</span><span id="line-182"> private Pair&lt;FSDataInputStream, FileStatus&gt; openArchivedWAL() throws IOException {</span>
<span class="source-line-no">183</span><span id="line-183"> Path archivedWAL = AbstractFSWALProvider.findArchivedLog(path, conf);</span>
<span class="source-line-no">184</span><span id="line-184"> if (archivedWAL != null) {</span>
<span class="source-line-no">185</span><span id="line-185"> // try open from oldWAL dir</span>
<span class="source-line-no">186</span><span id="line-186"> return Pair.newPair(fs.open(archivedWAL), fs.getFileStatus(archivedWAL));</span>
<span class="source-line-no">187</span><span id="line-187"> } else {</span>
<span class="source-line-no">188</span><span id="line-188"> return null;</span>
<span class="source-line-no">189</span><span id="line-189"> }</span>
<span class="source-line-no">190</span><span id="line-190"> }</span>
<span class="source-line-no">191</span><span id="line-191"></span>
<span class="source-line-no">192</span><span id="line-192"> protected final Pair&lt;FSDataInputStream, FileStatus&gt; open() throws IOException {</span>
<span class="source-line-no">193</span><span id="line-193"> try {</span>
<span class="source-line-no">194</span><span id="line-194"> return Pair.newPair(fs.open(path), fs.getFileStatus(path));</span>
<span class="source-line-no">195</span><span id="line-195"> } catch (FileNotFoundException e) {</span>
<span class="source-line-no">196</span><span id="line-196"> Pair&lt;FSDataInputStream, FileStatus&gt; pair = openArchivedWAL();</span>
<span class="source-line-no">197</span><span id="line-197"> if (pair != null) {</span>
<span class="source-line-no">198</span><span id="line-198"> return pair;</span>
<span class="source-line-no">199</span><span id="line-199"> } else {</span>
<span class="source-line-no">200</span><span id="line-200"> throw e;</span>
<span class="source-line-no">201</span><span id="line-201"> }</span>
<span class="source-line-no">202</span><span id="line-202"> } catch (RemoteException re) {</span>
<span class="source-line-no">203</span><span id="line-203"> IOException ioe = re.unwrapRemoteException(FileNotFoundException.class);</span>
<span class="source-line-no">204</span><span id="line-204"> if (!(ioe instanceof FileNotFoundException)) {</span>
<span class="source-line-no">205</span><span id="line-205"> throw ioe;</span>
<span class="source-line-no">206</span><span id="line-206"> }</span>
<span class="source-line-no">207</span><span id="line-207"> Pair&lt;FSDataInputStream, FileStatus&gt; pair = openArchivedWAL();</span>
<span class="source-line-no">208</span><span id="line-208"> if (pair != null) {</span>
<span class="source-line-no">209</span><span id="line-209"> return pair;</span>
<span class="source-line-no">210</span><span id="line-210"> } else {</span>
<span class="source-line-no">211</span><span id="line-211"> throw ioe;</span>
<span class="source-line-no">212</span><span id="line-212"> }</span>
<span class="source-line-no">213</span><span id="line-213"> }</span>
<span class="source-line-no">214</span><span id="line-214"> }</span>
<span class="source-line-no">215</span><span id="line-215"></span>
<span class="source-line-no">216</span><span id="line-216"> protected final WALProtos.WALHeader readHeader(FSDataInputStream stream) throws IOException {</span>
<span class="source-line-no">217</span><span id="line-217"> byte[] magic = new byte[PB_WAL_MAGIC.length];</span>
<span class="source-line-no">218</span><span id="line-218"> try {</span>
<span class="source-line-no">219</span><span id="line-219"> stream.readFully(magic);</span>
<span class="source-line-no">220</span><span id="line-220"> } catch (EOFException e) {</span>
<span class="source-line-no">221</span><span id="line-221"> throw new WALHeaderEOFException("EOF while reading PB WAL magic", e);</span>
<span class="source-line-no">222</span><span id="line-222"> }</span>
<span class="source-line-no">223</span><span id="line-223"> if (!Arrays.equals(PB_WAL_MAGIC, magic)) {</span>
<span class="source-line-no">224</span><span id="line-224"> throw new IOException("Invalid PB WAL magic " + Bytes.toStringBinary(magic) + ", expected "</span>
<span class="source-line-no">225</span><span id="line-225"> + Bytes.toStringBinary(PB_WAL_MAGIC));</span>
<span class="source-line-no">226</span><span id="line-226"> }</span>
<span class="source-line-no">227</span><span id="line-227"> WALProtos.WALHeader header;</span>
<span class="source-line-no">228</span><span id="line-228"> try {</span>
<span class="source-line-no">229</span><span id="line-229"> header = ProtobufUtil.parseDelimitedFrom(stream, WALProtos.WALHeader.parser());</span>
<span class="source-line-no">230</span><span id="line-230"> } catch (InvalidProtocolBufferException e) {</span>
<span class="source-line-no">231</span><span id="line-231"> if (ProtobufUtil.isEOF(e)) {</span>
<span class="source-line-no">232</span><span id="line-232"> throw new WALHeaderEOFException("EOF while reading PB header", e);</span>
<span class="source-line-no">233</span><span id="line-233"> } else {</span>
<span class="source-line-no">234</span><span id="line-234"> throw e;</span>
<span class="source-line-no">235</span><span id="line-235"> }</span>
<span class="source-line-no">236</span><span id="line-236"> } catch (EOFException e) {</span>
<span class="source-line-no">237</span><span id="line-237"> throw new WALHeaderEOFException("EOF while reading PB header", e);</span>
<span class="source-line-no">238</span><span id="line-238"> }</span>
<span class="source-line-no">239</span><span id="line-239"> if (header == null) {</span>
<span class="source-line-no">240</span><span id="line-240"> throw new WALHeaderEOFException("EOF while reading PB header");</span>
<span class="source-line-no">241</span><span id="line-241"> }</span>
<span class="source-line-no">242</span><span id="line-242"> if (header.hasWriterClsName() &amp;&amp; !getWriterClsNames().contains(header.getWriterClsName())) {</span>
<span class="source-line-no">243</span><span id="line-243"> throw new IOException("Got unknown writer class: " + header.getWriterClsName());</span>
<span class="source-line-no">244</span><span id="line-244"> }</span>
<span class="source-line-no">245</span><span id="line-245"> return header;</span>
<span class="source-line-no">246</span><span id="line-246"> }</span>
<span class="source-line-no">247</span><span id="line-247"></span>
<span class="source-line-no">248</span><span id="line-248"> private void initDecryptor(WALProtos.WALHeader header) throws IOException {</span>
<span class="source-line-no">249</span><span id="line-249"> if (!header.hasEncryptionKey()) {</span>
<span class="source-line-no">250</span><span id="line-250"> return;</span>
<span class="source-line-no">251</span><span id="line-251"> }</span>
<span class="source-line-no">252</span><span id="line-252"> EncryptionTest.testKeyProvider(conf);</span>
<span class="source-line-no">253</span><span id="line-253"> EncryptionTest.testCipherProvider(conf);</span>
<span class="source-line-no">254</span><span id="line-254"></span>
<span class="source-line-no">255</span><span id="line-255"> // Retrieve a usable key</span>
<span class="source-line-no">256</span><span id="line-256"> byte[] keyBytes = header.getEncryptionKey().toByteArray();</span>
<span class="source-line-no">257</span><span id="line-257"> Key key = null;</span>
<span class="source-line-no">258</span><span id="line-258"> String walKeyName = conf.get(HConstants.CRYPTO_WAL_KEY_NAME_CONF_KEY);</span>
<span class="source-line-no">259</span><span id="line-259"> // First try the WAL key, if one is configured</span>
<span class="source-line-no">260</span><span id="line-260"> if (walKeyName != null) {</span>
<span class="source-line-no">261</span><span id="line-261"> try {</span>
<span class="source-line-no">262</span><span id="line-262"> key = EncryptionUtil.unwrapWALKey(conf, walKeyName, keyBytes);</span>
<span class="source-line-no">263</span><span id="line-263"> } catch (KeyException e) {</span>
<span class="source-line-no">264</span><span id="line-264"> LOG.debug("Unable to unwrap key with WAL key '{}'", walKeyName, e);</span>
<span class="source-line-no">265</span><span id="line-265"> key = null;</span>
<span class="source-line-no">266</span><span id="line-266"> }</span>
<span class="source-line-no">267</span><span id="line-267"> }</span>
<span class="source-line-no">268</span><span id="line-268"> if (key == null) {</span>
<span class="source-line-no">269</span><span id="line-269"> String masterKeyName =</span>
<span class="source-line-no">270</span><span id="line-270"> conf.get(HConstants.CRYPTO_MASTERKEY_NAME_CONF_KEY, User.getCurrent().getShortName());</span>
<span class="source-line-no">271</span><span id="line-271"> try {</span>
<span class="source-line-no">272</span><span id="line-272"> // Then, try the cluster master key</span>
<span class="source-line-no">273</span><span id="line-273"> key = EncryptionUtil.unwrapWALKey(conf, masterKeyName, keyBytes);</span>
<span class="source-line-no">274</span><span id="line-274"> } catch (KeyException e) {</span>
<span class="source-line-no">275</span><span id="line-275"> // If the current master key fails to unwrap, try the alternate, if</span>
<span class="source-line-no">276</span><span id="line-276"> // one is configured</span>
<span class="source-line-no">277</span><span id="line-277"> LOG.debug("Unable to unwrap key with current master key '{}'", masterKeyName, e);</span>
<span class="source-line-no">278</span><span id="line-278"> String alternateKeyName = conf.get(HConstants.CRYPTO_MASTERKEY_ALTERNATE_NAME_CONF_KEY);</span>
<span class="source-line-no">279</span><span id="line-279"> if (alternateKeyName != null) {</span>
<span class="source-line-no">280</span><span id="line-280"> try {</span>
<span class="source-line-no">281</span><span id="line-281"> key = EncryptionUtil.unwrapWALKey(conf, alternateKeyName, keyBytes);</span>
<span class="source-line-no">282</span><span id="line-282"> } catch (KeyException ex) {</span>
<span class="source-line-no">283</span><span id="line-283"> throw new IOException(ex);</span>
<span class="source-line-no">284</span><span id="line-284"> }</span>
<span class="source-line-no">285</span><span id="line-285"> } else {</span>
<span class="source-line-no">286</span><span id="line-286"> throw new IOException(e);</span>
<span class="source-line-no">287</span><span id="line-287"> }</span>
<span class="source-line-no">288</span><span id="line-288"> }</span>
<span class="source-line-no">289</span><span id="line-289"> }</span>
<span class="source-line-no">290</span><span id="line-290"></span>
<span class="source-line-no">291</span><span id="line-291"> // Use the algorithm the key wants</span>
<span class="source-line-no">292</span><span id="line-292"> Cipher cipher = Encryption.getCipher(conf, key.getAlgorithm());</span>
<span class="source-line-no">293</span><span id="line-293"> if (cipher == null) {</span>
<span class="source-line-no">294</span><span id="line-294"> throw new IOException("Cipher '" + key.getAlgorithm() + "' is not available");</span>
<span class="source-line-no">295</span><span id="line-295"> }</span>
<span class="source-line-no">296</span><span id="line-296"></span>
<span class="source-line-no">297</span><span id="line-297"> // Set up the decryptor for this WAL</span>
<span class="source-line-no">298</span><span id="line-298"></span>
<span class="source-line-no">299</span><span id="line-299"> decryptor = cipher.getDecryptor();</span>
<span class="source-line-no">300</span><span id="line-300"> decryptor.setKey(key);</span>
<span class="source-line-no">301</span><span id="line-301"></span>
<span class="source-line-no">302</span><span id="line-302"> LOG.debug("Initialized secure protobuf WAL: cipher={}", cipher.getName());</span>
<span class="source-line-no">303</span><span id="line-303"> }</span>
<span class="source-line-no">304</span><span id="line-304"></span>
<span class="source-line-no">305</span><span id="line-305"> private void initCompression(WALProtos.WALHeader header) throws IOException {</span>
<span class="source-line-no">306</span><span id="line-306"> this.hasCompression = header.hasHasCompression() &amp;&amp; header.getHasCompression();</span>
<span class="source-line-no">307</span><span id="line-307"> if (!hasCompression) {</span>
<span class="source-line-no">308</span><span id="line-308"> return;</span>
<span class="source-line-no">309</span><span id="line-309"> }</span>
<span class="source-line-no">310</span><span id="line-310"> this.hasTagCompression = header.hasHasTagCompression() &amp;&amp; header.getHasTagCompression();</span>
<span class="source-line-no">311</span><span id="line-311"> this.hasValueCompression = header.hasHasValueCompression() &amp;&amp; header.getHasValueCompression();</span>
<span class="source-line-no">312</span><span id="line-312"> if (header.hasValueCompressionAlgorithm()) {</span>
<span class="source-line-no">313</span><span id="line-313"> try {</span>
<span class="source-line-no">314</span><span id="line-314"> this.valueCompressionType =</span>
<span class="source-line-no">315</span><span id="line-315"> Compression.Algorithm.values()[header.getValueCompressionAlgorithm()];</span>
<span class="source-line-no">316</span><span id="line-316"> } catch (ArrayIndexOutOfBoundsException e) {</span>
<span class="source-line-no">317</span><span id="line-317"> throw new IOException("Invalid compression type", e);</span>
<span class="source-line-no">318</span><span id="line-318"> }</span>
<span class="source-line-no">319</span><span id="line-319"> }</span>
<span class="source-line-no">320</span><span id="line-320"> if (LOG.isDebugEnabled()) {</span>
<span class="source-line-no">321</span><span id="line-321"> LOG.debug(</span>
<span class="source-line-no">322</span><span id="line-322"> "Initializing compression context for {}: isRecoveredEdits={}"</span>
<span class="source-line-no">323</span><span id="line-323"> + ", hasTagCompression={}, hasValueCompression={}, valueCompressionType={}",</span>
<span class="source-line-no">324</span><span id="line-324"> path, CommonFSUtils.isRecoveredEdits(path), hasTagCompression, hasValueCompression,</span>
<span class="source-line-no">325</span><span id="line-325"> valueCompressionType);</span>
<span class="source-line-no">326</span><span id="line-326"> }</span>
<span class="source-line-no">327</span><span id="line-327"> try {</span>
<span class="source-line-no">328</span><span id="line-328"> compressionCtx =</span>
<span class="source-line-no">329</span><span id="line-329"> new CompressionContext(LRUDictionary.class, CommonFSUtils.isRecoveredEdits(path),</span>
<span class="source-line-no">330</span><span id="line-330"> hasTagCompression, hasValueCompression, valueCompressionType);</span>
<span class="source-line-no">331</span><span id="line-331"> } catch (Exception e) {</span>
<span class="source-line-no">332</span><span id="line-332"> throw new IOException("Failed to initialize CompressionContext", e);</span>
<span class="source-line-no">333</span><span id="line-333"> }</span>
<span class="source-line-no">334</span><span id="line-334"> }</span>
<span class="source-line-no">335</span><span id="line-335"></span>
<span class="source-line-no">336</span><span id="line-336"> private WALCellCodec getCodec(Configuration conf, String cellCodecClsName,</span>
<span class="source-line-no">337</span><span id="line-337"> CompressionContext compressionContext) throws IOException {</span>
<span class="source-line-no">338</span><span id="line-338"> return WALCellCodec.create(conf, cellCodecClsName, compressionContext);</span>
<span class="source-line-no">339</span><span id="line-339"> }</span>
<span class="source-line-no">340</span><span id="line-340"></span>
<span class="source-line-no">341</span><span id="line-341"> protected final void initWALCellCodec(WALProtos.WALHeader header, InputStream inputStream)</span>
<span class="source-line-no">342</span><span id="line-342"> throws IOException {</span>
<span class="source-line-no">343</span><span id="line-343"> String cellCodecClsName = header.hasCellCodecClsName() ? header.getCellCodecClsName() : null;</span>
<span class="source-line-no">344</span><span id="line-344"> if (decryptor != null &amp;&amp; SecureWALCellCodec.class.getName().equals(cellCodecClsName)) {</span>
<span class="source-line-no">345</span><span id="line-345"> WALCellCodec codec = SecureWALCellCodec.getCodec(this.conf, decryptor);</span>
<span class="source-line-no">346</span><span id="line-346"> this.cellDecoder = codec.getDecoder(inputStream);</span>
<span class="source-line-no">347</span><span id="line-347"> // We do not support compression with WAL encryption</span>
<span class="source-line-no">348</span><span id="line-348"> this.compressionCtx = null;</span>
<span class="source-line-no">349</span><span id="line-349"> this.byteStringUncompressor = WALCellCodec.getNoneUncompressor();</span>
<span class="source-line-no">350</span><span id="line-350"> this.hasCompression = false;</span>
<span class="source-line-no">351</span><span id="line-351"> this.hasTagCompression = false;</span>
<span class="source-line-no">352</span><span id="line-352"> this.hasValueCompression = false;</span>
<span class="source-line-no">353</span><span id="line-353"> } else {</span>
<span class="source-line-no">354</span><span id="line-354"> WALCellCodec codec = getCodec(conf, cellCodecClsName, compressionCtx);</span>
<span class="source-line-no">355</span><span id="line-355"> this.cellDecoder = codec.getDecoder(inputStream);</span>
<span class="source-line-no">356</span><span id="line-356"> if (this.hasCompression) {</span>
<span class="source-line-no">357</span><span id="line-357"> this.byteStringUncompressor = codec.getByteStringUncompressor();</span>
<span class="source-line-no">358</span><span id="line-358"> } else {</span>
<span class="source-line-no">359</span><span id="line-359"> this.byteStringUncompressor = WALCellCodec.getNoneUncompressor();</span>
<span class="source-line-no">360</span><span id="line-360"> }</span>
<span class="source-line-no">361</span><span id="line-361"> }</span>
<span class="source-line-no">362</span><span id="line-362"> this.codecClsName = cellCodecClsName;</span>
<span class="source-line-no">363</span><span id="line-363"> }</span>
<span class="source-line-no">364</span><span id="line-364"></span>
<span class="source-line-no">365</span><span id="line-365"> protected final void readTrailer(FSDataInputStream stream, FileStatus stat) throws IOException {</span>
<span class="source-line-no">366</span><span id="line-366"> this.fileLength = stat.getLen();</span>
<span class="source-line-no">367</span><span id="line-367"> this.walEditsStopOffset = this.fileLength;</span>
<span class="source-line-no">368</span><span id="line-368"> long currentPos = stream.getPos();</span>
<span class="source-line-no">369</span><span id="line-369"> // we will reset walEditsStopOffset if trailer is available</span>
<span class="source-line-no">370</span><span id="line-370"> trailerPresent = setTrailerIfPresent(stream);</span>
<span class="source-line-no">371</span><span id="line-371"> if (currentPos != stream.getPos()) {</span>
<span class="source-line-no">372</span><span id="line-372"> // seek back</span>
<span class="source-line-no">373</span><span id="line-373"> stream.seek(currentPos);</span>
<span class="source-line-no">374</span><span id="line-374"> }</span>
<span class="source-line-no">375</span><span id="line-375"> }</span>
<span class="source-line-no">376</span><span id="line-376"></span>
<span class="source-line-no">377</span><span id="line-377"> /**</span>
<span class="source-line-no">378</span><span id="line-378"> * To check whether a trailer is present in a WAL, it seeks to position (fileLength -</span>
<span class="source-line-no">379</span><span id="line-379"> * PB_WAL_COMPLETE_MAGIC.size() - Bytes.SIZEOF_INT). It reads the int value to know the size of</span>
<span class="source-line-no">380</span><span id="line-380"> * the trailer, and checks whether the trailer is present at the end or not by comparing the last</span>
<span class="source-line-no">381</span><span id="line-381"> * PB_WAL_COMPLETE_MAGIC.size() bytes. In case trailer is not present, it returns false;</span>
<span class="source-line-no">382</span><span id="line-382"> * otherwise, sets the trailer and sets this.walEditsStopOffset variable up to the point just</span>
<span class="source-line-no">383</span><span id="line-383"> * before the trailer.</span>
<span class="source-line-no">384</span><span id="line-384"> * &lt;p/&gt;</span>
<span class="source-line-no">385</span><span id="line-385"> * The trailer is ignored in case:</span>
<span class="source-line-no">386</span><span id="line-386"> * &lt;ul&gt;</span>
<span class="source-line-no">387</span><span id="line-387"> * &lt;li&gt;fileLength is 0 or not correct (when file is under recovery, etc).</span>
<span class="source-line-no">388</span><span id="line-388"> * &lt;li&gt;the trailer size is negative.</span>
<span class="source-line-no">389</span><span id="line-389"> * &lt;/ul&gt;</span>
<span class="source-line-no">390</span><span id="line-390"> * In case the trailer size &gt; this.trailerMaxSize, it is read after a WARN message.</span>
<span class="source-line-no">391</span><span id="line-391"> * @return true if a valid trailer is present</span>
<span class="source-line-no">392</span><span id="line-392"> */</span>
<span class="source-line-no">393</span><span id="line-393"> private boolean setTrailerIfPresent(FSDataInputStream stream) throws IOException {</span>
<span class="source-line-no">394</span><span id="line-394"> try {</span>
<span class="source-line-no">395</span><span id="line-395"> long trailerSizeOffset = this.fileLength - (PB_WAL_COMPLETE_MAGIC.length + Bytes.SIZEOF_INT);</span>
<span class="source-line-no">396</span><span id="line-396"> if (trailerSizeOffset &lt;= 0) {</span>
<span class="source-line-no">397</span><span id="line-397"> // no trailer possible.</span>
<span class="source-line-no">398</span><span id="line-398"> return false;</span>
<span class="source-line-no">399</span><span id="line-399"> }</span>
<span class="source-line-no">400</span><span id="line-400"> stream.seek(trailerSizeOffset);</span>
<span class="source-line-no">401</span><span id="line-401"> // read the int as trailer size.</span>
<span class="source-line-no">402</span><span id="line-402"> int trailerSize = stream.readInt();</span>
<span class="source-line-no">403</span><span id="line-403"> ByteBuffer buf = ByteBuffer.allocate(PB_WAL_COMPLETE_MAGIC.length);</span>
<span class="source-line-no">404</span><span id="line-404"> stream.readFully(buf.array(), buf.arrayOffset(), buf.capacity());</span>
<span class="source-line-no">405</span><span id="line-405"> if (!Arrays.equals(buf.array(), PB_WAL_COMPLETE_MAGIC)) {</span>
<span class="source-line-no">406</span><span id="line-406"> LOG.trace("No trailer found.");</span>
<span class="source-line-no">407</span><span id="line-407"> return false;</span>
<span class="source-line-no">408</span><span id="line-408"> }</span>
<span class="source-line-no">409</span><span id="line-409"> if (trailerSize &lt; 0) {</span>
<span class="source-line-no">410</span><span id="line-410"> LOG.warn("Invalid trailer Size " + trailerSize + ", ignoring the trailer");</span>
<span class="source-line-no">411</span><span id="line-411"> return false;</span>
<span class="source-line-no">412</span><span id="line-412"> } else if (trailerSize &gt; this.trailerWarnSize) {</span>
<span class="source-line-no">413</span><span id="line-413"> // continue reading after warning the user.</span>
<span class="source-line-no">414</span><span id="line-414"> LOG.warn("Please investigate WALTrailer usage. Trailer size &gt; maximum configured size : "</span>
<span class="source-line-no">415</span><span id="line-415"> + trailerSize + " &gt; " + this.trailerWarnSize);</span>
<span class="source-line-no">416</span><span id="line-416"> }</span>
<span class="source-line-no">417</span><span id="line-417"> // seek to the position where trailer starts.</span>
<span class="source-line-no">418</span><span id="line-418"> long positionOfTrailer = trailerSizeOffset - trailerSize;</span>
<span class="source-line-no">419</span><span id="line-419"> stream.seek(positionOfTrailer);</span>
<span class="source-line-no">420</span><span id="line-420"> // read the trailer.</span>
<span class="source-line-no">421</span><span id="line-421"> buf = ByteBuffer.allocate(trailerSize);// for trailer.</span>
<span class="source-line-no">422</span><span id="line-422"> stream.readFully(buf.array(), buf.arrayOffset(), buf.capacity());</span>
<span class="source-line-no">423</span><span id="line-423"> trailer = WALTrailer.parseFrom(buf.array());</span>
<span class="source-line-no">424</span><span id="line-424"> this.walEditsStopOffset = positionOfTrailer;</span>
<span class="source-line-no">425</span><span id="line-425"> return true;</span>
<span class="source-line-no">426</span><span id="line-426"> } catch (IOException ioe) {</span>
<span class="source-line-no">427</span><span id="line-427"> LOG.warn("Got IOE while reading the trailer. Continuing as if no trailer is present.", ioe);</span>
<span class="source-line-no">428</span><span id="line-428"> }</span>
<span class="source-line-no">429</span><span id="line-429"> return false;</span>
<span class="source-line-no">430</span><span id="line-430"> }</span>
<span class="source-line-no">431</span><span id="line-431"></span>
<span class="source-line-no">432</span><span id="line-432"> protected final boolean reachWALEditsStopOffset(long pos) {</span>
<span class="source-line-no">433</span><span id="line-433"> if (trailerPresent &amp;&amp; pos &gt; 0 &amp;&amp; pos == walEditsStopOffset) {</span>
<span class="source-line-no">434</span><span id="line-434"> LOG.trace("Reached end of expected edits area at offset {}", pos);</span>
<span class="source-line-no">435</span><span id="line-435"> return true;</span>
<span class="source-line-no">436</span><span id="line-436"> } else {</span>
<span class="source-line-no">437</span><span id="line-437"> return false;</span>
<span class="source-line-no">438</span><span id="line-438"> }</span>
<span class="source-line-no">439</span><span id="line-439"> }</span>
<span class="source-line-no">440</span><span id="line-440"></span>
<span class="source-line-no">441</span><span id="line-441"> /**</span>
<span class="source-line-no">442</span><span id="line-442"> * Returns names of the accepted writer classes</span>
<span class="source-line-no">443</span><span id="line-443"> */</span>
<span class="source-line-no">444</span><span id="line-444"> public List&lt;String&gt; getWriterClsNames() {</span>
<span class="source-line-no">445</span><span id="line-445"> return WRITER_CLS_NAMES;</span>
<span class="source-line-no">446</span><span id="line-446"> }</span>
<span class="source-line-no">447</span><span id="line-447"></span>
<span class="source-line-no">448</span><span id="line-448"> /**</span>
<span class="source-line-no">449</span><span id="line-449"> * Returns the cell codec classname</span>
<span class="source-line-no">450</span><span id="line-450"> */</span>
<span class="source-line-no">451</span><span id="line-451"> public String getCodecClsName() {</span>
<span class="source-line-no">452</span><span id="line-452"> return codecClsName;</span>
<span class="source-line-no">453</span><span id="line-453"> }</span>
<span class="source-line-no">454</span><span id="line-454"></span>
<span class="source-line-no">455</span><span id="line-455"> public long getPosition() throws IOException {</span>
<span class="source-line-no">456</span><span id="line-456"> return inputStream != null ? inputStream.getPos() : -1;</span>
<span class="source-line-no">457</span><span id="line-457"> }</span>
<span class="source-line-no">458</span><span id="line-458"></span>
<span class="source-line-no">459</span><span id="line-459"> public long trailerSize() {</span>
<span class="source-line-no">460</span><span id="line-460"> if (trailerPresent) {</span>
<span class="source-line-no">461</span><span id="line-461"> // sizeof PB_WAL_COMPLETE_MAGIC + sizeof trailerSize + trailer</span>
<span class="source-line-no">462</span><span id="line-462"> final long calculatedSize =</span>
<span class="source-line-no">463</span><span id="line-463"> (long) PB_WAL_COMPLETE_MAGIC.length + Bytes.SIZEOF_INT + trailer.getSerializedSize();</span>
<span class="source-line-no">464</span><span id="line-464"> final long expectedSize = fileLength - walEditsStopOffset;</span>
<span class="source-line-no">465</span><span id="line-465"> if (expectedSize != calculatedSize) {</span>
<span class="source-line-no">466</span><span id="line-466"> LOG.warn("After parsing the trailer, we expect the total footer to be {} bytes, but we "</span>
<span class="source-line-no">467</span><span id="line-467"> + "calculate it as being {}", expectedSize, calculatedSize);</span>
<span class="source-line-no">468</span><span id="line-468"> }</span>
<span class="source-line-no">469</span><span id="line-469"> return expectedSize;</span>
<span class="source-line-no">470</span><span id="line-470"> } else {</span>
<span class="source-line-no">471</span><span id="line-471"> return -1L;</span>
<span class="source-line-no">472</span><span id="line-472"> }</span>
<span class="source-line-no">473</span><span id="line-473"> }</span>
<span class="source-line-no">474</span><span id="line-474"></span>
<span class="source-line-no">475</span><span id="line-475"> protected final String getPositionQuietly() {</span>
<span class="source-line-no">476</span><span id="line-476"> try {</span>
<span class="source-line-no">477</span><span id="line-477"> long pos = getPosition();</span>
<span class="source-line-no">478</span><span id="line-478"> return pos &gt;= 0 ? Long.toString(pos) : "&lt;unknown&gt;";</span>
<span class="source-line-no">479</span><span id="line-479"> } catch (Exception e) {</span>
<span class="source-line-no">480</span><span id="line-480"> LOG.warn("failed to get position, ignoring", e);</span>
<span class="source-line-no">481</span><span id="line-481"> return "&lt;unknown&gt;";</span>
<span class="source-line-no">482</span><span id="line-482"> }</span>
<span class="source-line-no">483</span><span id="line-483"> }</span>
<span class="source-line-no">484</span><span id="line-484"></span>
<span class="source-line-no">485</span><span id="line-485"> protected final IOException extractHiddenEof(Exception ex) {</span>
<span class="source-line-no">486</span><span id="line-486"> // There are two problems we are dealing with here. Hadoop stream throws generic exception</span>
<span class="source-line-no">487</span><span id="line-487"> // for EOF, not EOFException; and scanner further hides it inside RuntimeException.</span>
<span class="source-line-no">488</span><span id="line-488"> IOException ioEx = null;</span>
<span class="source-line-no">489</span><span id="line-489"> if (ex instanceof EOFException) {</span>
<span class="source-line-no">490</span><span id="line-490"> return (EOFException) ex;</span>
<span class="source-line-no">491</span><span id="line-491"> } else if (ex instanceof IOException) {</span>
<span class="source-line-no">492</span><span id="line-492"> ioEx = (IOException) ex;</span>
<span class="source-line-no">493</span><span id="line-493"> } else if (</span>
<span class="source-line-no">494</span><span id="line-494"> ex instanceof RuntimeException &amp;&amp; ex.getCause() != null</span>
<span class="source-line-no">495</span><span id="line-495"> &amp;&amp; ex.getCause() instanceof IOException</span>
<span class="source-line-no">496</span><span id="line-496"> ) {</span>
<span class="source-line-no">497</span><span id="line-497"> ioEx = (IOException) ex.getCause();</span>
<span class="source-line-no">498</span><span id="line-498"> }</span>
<span class="source-line-no">499</span><span id="line-499"> if ((ioEx != null) &amp;&amp; (ioEx.getMessage() != null)) {</span>
<span class="source-line-no">500</span><span id="line-500"> if (ioEx.getMessage().contains("EOF")) {</span>
<span class="source-line-no">501</span><span id="line-501"> return ioEx;</span>
<span class="source-line-no">502</span><span id="line-502"> }</span>
<span class="source-line-no">503</span><span id="line-503"> return null;</span>
<span class="source-line-no">504</span><span id="line-504"> }</span>
<span class="source-line-no">505</span><span id="line-505"> return null;</span>
<span class="source-line-no">506</span><span id="line-506"> }</span>
<span class="source-line-no">507</span><span id="line-507"></span>
<span class="source-line-no">508</span><span id="line-508"> /**</span>
<span class="source-line-no">509</span><span id="line-509"> * This is used to determine whether we have already reached the WALTrailer. As the size and magic</span>
<span class="source-line-no">510</span><span id="line-510"> * are at the end of the WAL file, it is possible that these two options are missing while</span>
<span class="source-line-no">511</span><span id="line-511"> * writing, so we will consider there is no trailer. And when we actually reach the WALTrailer, we</span>
<span class="source-line-no">512</span><span id="line-512"> * will try to decode it as WALKey and we will fail but the error could be varied as it is parsing</span>
<span class="source-line-no">513</span><span id="line-513"> * WALTrailer actually.</span>
<span class="source-line-no">514</span><span id="line-514"> * @return whether this is a WALTrailer and we should throw EOF to upper layer the file is done</span>
<span class="source-line-no">515</span><span id="line-515"> */</span>
<span class="source-line-no">516</span><span id="line-516"> protected final boolean isWALTrailer(long startPosition) throws IOException {</span>
<span class="source-line-no">517</span><span id="line-517"> // We have nothing in the WALTrailer PB message now so its size is just an int length size and a</span>
<span class="source-line-no">518</span><span id="line-518"> // magic at the end</span>
<span class="source-line-no">519</span><span id="line-519"> int trailerSize = PB_WAL_COMPLETE_MAGIC.length + Bytes.SIZEOF_INT;</span>
<span class="source-line-no">520</span><span id="line-520"> if (fileLength - startPosition &gt;= trailerSize) {</span>
<span class="source-line-no">521</span><span id="line-521"> // We still have more than trailerSize bytes before reaching the EOF so this is not a trailer.</span>
<span class="source-line-no">522</span><span id="line-522"> // We also test for == here because if this is a valid trailer, we can read it while opening</span>
<span class="source-line-no">523</span><span id="line-523"> // the reader, so we should not reach here</span>
<span class="source-line-no">524</span><span id="line-524"> return false;</span>
<span class="source-line-no">525</span><span id="line-525"> }</span>
<span class="source-line-no">526</span><span id="line-526"> inputStream.seek(startPosition);</span>
<span class="source-line-no">527</span><span id="line-527"> for (int i = 0; i &lt; 4; i++) {</span>
<span class="source-line-no">528</span><span id="line-528"> int r = inputStream.read();</span>
<span class="source-line-no">529</span><span id="line-529"> if (r == -1) {</span>
<span class="source-line-no">530</span><span id="line-530"> // we have reached EOF while reading the length, and all bytes read are 0, so we assume this</span>
<span class="source-line-no">531</span><span id="line-531"> // is a partial trailer</span>
<span class="source-line-no">532</span><span id="line-532"> return true;</span>
<span class="source-line-no">533</span><span id="line-533"> }</span>
<span class="source-line-no">534</span><span id="line-534"> if (r != 0) {</span>
<span class="source-line-no">535</span><span id="line-535"> // the length is not 0, should not be a trailer</span>
<span class="source-line-no">536</span><span id="line-536"> return false;</span>
<span class="source-line-no">537</span><span id="line-537"> }</span>
<span class="source-line-no">538</span><span id="line-538"> }</span>
<span class="source-line-no">539</span><span id="line-539"> for (int i = 0; i &lt; PB_WAL_COMPLETE_MAGIC.length; i++) {</span>
<span class="source-line-no">540</span><span id="line-540"> int r = inputStream.read();</span>
<span class="source-line-no">541</span><span id="line-541"> if (r == -1) {</span>
<span class="source-line-no">542</span><span id="line-542"> // we have reached EOF while reading the magic, and all bytes read are matched, so we assume</span>
<span class="source-line-no">543</span><span id="line-543"> // this is a partial trailer</span>
<span class="source-line-no">544</span><span id="line-544"> return true;</span>
<span class="source-line-no">545</span><span id="line-545"> }</span>
<span class="source-line-no">546</span><span id="line-546"> if (r != (PB_WAL_COMPLETE_MAGIC[i] &amp; 0xFF)) {</span>
<span class="source-line-no">547</span><span id="line-547"> // does not match magic, should not be a trailer</span>
<span class="source-line-no">548</span><span id="line-548"> return false;</span>
<span class="source-line-no">549</span><span id="line-549"> }</span>
<span class="source-line-no">550</span><span id="line-550"> }</span>
<span class="source-line-no">551</span><span id="line-551"> // in fact, we should not reach here, as this means the trailer bytes are all matched and</span>
<span class="source-line-no">552</span><span id="line-552"> // complete, then we should not call this method...</span>
<span class="source-line-no">553</span><span id="line-553"> return true;</span>
<span class="source-line-no">554</span><span id="line-554"> }</span>
<span class="source-line-no">555</span><span id="line-555"></span>
<span class="source-line-no">556</span><span id="line-556"> @Override</span>
<span class="source-line-no">557</span><span id="line-557"> public void close() {</span>
<span class="source-line-no">558</span><span id="line-558"> if (inputStream != null) {</span>
<span class="source-line-no">559</span><span id="line-559"> IOUtils.closeQuietly(inputStream);</span>
<span class="source-line-no">560</span><span id="line-560"> inputStream = null;</span>
<span class="source-line-no">561</span><span id="line-561"> }</span>
<span class="source-line-no">562</span><span id="line-562"> }</span>
<span class="source-line-no">563</span><span id="line-563">}</span>
</pre>
</div>
</main>
</body>
</html>