bbeaudreault commented on a change in pull request #3803: URL: https://github.com/apache/hbase/pull/3803#discussion_r752631908
########## File path: hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/InputStreamBlockDistribution.java ########## @@ -0,0 +1,143 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hbase.regionserver; + +import com.google.errorprone.annotations.RestrictedApi; +import java.io.IOException; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.hbase.HDFSBlocksDistribution; +import org.apache.hadoop.hbase.io.FileLink; +import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; +import org.apache.hadoop.hbase.util.FSUtils; +import org.apache.hadoop.hdfs.client.HdfsDataInputStream; +import org.apache.yetus.audience.InterfaceAudience; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Computes the HDFSBlockDistribution for a file based on the underlying located blocks + * for an HdfsDataInputStream reading that file. This computation may involve a call to + * the namenode, so the value is cached based on + * {@link #HBASE_LOCALITY_INPUTSTREAM_DERIVE_CACHE_PERIOD}. + */ [email protected] +public class InputStreamBlockDistribution { + private static final Logger LOG = LoggerFactory.getLogger(InputStreamBlockDistribution.class); + + private static final String HBASE_LOCALITY_INPUTSTREAM_DERIVE_ENABLED = + "hbase.locality.inputstream.derive.enabled"; + private static final boolean DEFAULT_HBASE_LOCALITY_INPUTSTREAM_DERIVE_ENABLED = false; + + private static final String HBASE_LOCALITY_INPUTSTREAM_DERIVE_CACHE_PERIOD = + "hbase.locality.inputstream.derive.cache.period"; + private static final int DEFAULT_HBASE_LOCALITY_INPUTSTREAM_DERIVE_CACHE_PERIOD = 60_000; + + private final FSDataInputStream stream; + private final StoreFileInfo fileInfo; + private final int cachePeriodMs; + + private HDFSBlocksDistribution hdfsBlocksDistribution; + private long lastCachedAt; + private boolean streamUnsupported; + + public InputStreamBlockDistribution(FSDataInputStream stream, StoreFileInfo fileInfo) { + this.stream = stream; Review comment: Thanks for reviewing, Duo! The stream is originally constructed here in [StoreFileInfo.createReaderContext]( https://github.com/apache/hbase/blob/master/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFileInfo.java#L293-L329). This method has two usages: - [HStoreFile.open](https://github.com/apache/hbase/blob/master/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStoreFile.java#L357), when opening a StoreFile and creating the `initialReader` which is used for all PREAD scans. - [HStoreFile.createStreamScanner](https://github.com/apache/hbase/blob/master/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStoreFile.java#L499), which is used for STREAM scans I'm interested in the first one, because it lives for the lifetime of a StoreFile. As far as I can tell, the initialReader (and thus this input stream) is never closed until the StoreFile itself closes. To your question, I don't think it makes sense to continue computing locality at that point. I don't see a case for the InputStream to be re-opened except through re-opening the StoreFile which would actually create a new FSDataInputStream and subsequently a new InputStreamBlockDistribution. The second one is relatively short lived, just for the lifetime of the scan. And nothing would be trying to fetch the locality from such a scan. So it doesn't make sense to worry too much about that one. --- Part of your question seems to be probing at whether this is too far removed from where the object is created. I could have considered putting all of the logic from this class into HStoreFile so it's clear that it's directly tied to that object's initialReader. However, HStoreFile is already a long class and IMO it's cleaner to put this nicely encapsulated logic in a separate class. This also makes it easier to test. I could also consider embedding this constructor call closer to where the InputStream itself is created, for example in [StoreFileInfo.createReaderContext]( https://github.com/apache/hbase/blob/master/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFileInfo.java#L293-L329), which returns an immutable `ReaderContext`. That could definitely be reasonable to do, but as mentioned above I really only want to do this for the StoreFile's primary initialReader... not for _every_ ReaderContext that is created. I could do it only when Reader.Type.PREAD is passed in, but that assume that we'll never create other PREAD ReaderContexts. So that's why I landed on instantiating this where I did, but definitely open to other opinions. --- As a final response to your direct question, once the InputStream is closed the calls to `DFSInputStream.getAllBlocks()` would theoretically still work, but I don't think they'd be getting updated overtime anymore. We probably also can't assume this will always be true, so best to assume that once the InputStream is closed the InputStreamBlockDistribution is defunct. I think this is totally fine in practice based on usage today, because the InputStream would only be closed when the StoreFile is closed and at that point nothing will be trying to fetch the locality info. I'd love to make this contract explicit with an `isClosed()` check, but FSDataInputStream does not expose that. Perhaps it makes sense to put this class directly into FSDataInputStreamWrapper and expose a new `getHDFSBlockDistribution()` method on that class. This way when FSDataInputStreamWrapper is closed, it's a bit more obvious that we shouldn't expect anything more from this class. My concern with this approach again goes back to the problem of embedding this in `createReaderContext` -- I don't really want to make every FSDataInputStreamWrapper have a computable locality, just the StoreFile's main one. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
