lee/heritrix3
1
0
Fork 0
mirror of https://github.com/internetarchive/heritrix3.git synced 2026-09-16 02:35:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove HBase from contrib

Browse Source 
The transitive dependencies pulled in via hbase are causing many security alerts for unpatched vulnerabilities in Dependabot, Maven Central and other tools. The newest version of hbase-client these modules seem to compile against still has many vulnerable transitive dependencies and the latest version of HBase appears to have many breaking API changes.

These can be re-added if someone volunteers to take over maintaining this code and updates it so that it doesn't pull in vulnerable dependencies.
This commit is contained in:
Alex Osborne
2024-10-25 14:33:26 +09:00
parent ff84c8d32f
commit 345e927d32
14 changed files with 0 additions and 1838 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contrib/pom.xml
-35
View File
@@ -24,41 +24,6 @@
<version>${project.version}</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>0.98.6-cdh5.3.5</version>
<exclusions>
<exclusion>
<artifactId>jets3t</artifactId>
<groupId>net.java.dev.jets3t</groupId>
</exclusion>
<exclusion>
<artifactId>junit</artifactId>
<groupId>junit</groupId>
</exclusion>
<!-- tools.jar is not available in JDK 11 so exclude it
hbase-client accidentally leaked it as a transitive dependency
https://issues.apache.org/jira/browse/HBASE-13963 -->
<exclusion>
<groupId>jdk.tools</groupId>
<artifactId>jdk.tools</artifactId>
</exclusion>
<!-- prefer guava 17 from webarchive-commons over guava 12 from hbase-client -->
<exclusion>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
</exclusion>
<exclusion>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
</exclusion>
<exclusion>
<groupId>commons-collections</groupId>
<artifactId>commons-collections</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
contrib/src/main/java/org/archive/modules/recrawl/FetchHistoryHelper.java
-32
View File
@@ -18,10 +18,6 @@
*/
package org.archive.modules.recrawl;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
@@ -82,35 +78,7 @@ public class FetchHistoryHelper {
return null;
}
protected static final DateFormat HTTP_DATE_FORMAT = new SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss zzz");
protected FetchHistoryHelper() {
}
/**
* converts time in HTTP Date format {@code dateStr} to seconds
* since epoch.
* @param dateStr time in HTTP Date format.
* @return seconds since epoch
*/
public static long parseHttpDate(String dateStr) {
synchronized (HTTP_DATE_FORMAT) {
try {
Date d = HTTP_DATE_FORMAT.parse(dateStr);
return d.getTime() / 1000;
} catch (ParseException ex) {
if (logger.isDebugEnabled())
logger.debug("bad HTTP DATE: " + dateStr);
return 0;
}
}
}
public static String formatHttpDate(long time) {
synchronized (HTTP_DATE_FORMAT) {
// format(Date) is not thread safe either
return HTTP_DATE_FORMAT.format(new Date(time * 1000));
}
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/HBase.java
-126
View File
@@ -1,126 +0,0 @@
/*
* This file is part of the Heritrix web crawler (crawler.archive.org).
*
* Licensed to the Internet Archive (IA) by one or more individual
* contributors.
*
* The IA 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.archive.modules.recrawl.hbase;
import java.io.IOException;
import java.util.Map;
import java.util.Map.Entry;
import java.util.logging.Logger;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.springframework.context.Lifecycle;
/**
* Represents a deployment of HBase. (An instance, a database, an HBase...)
*
* @author nlevitt
*/
public class HBase implements Lifecycle {
private static final Logger logger =
Logger.getLogger(HBase.class.getName());
protected Configuration conf = null;
private Map<String,String> properties;
public Map<String,String> getProperties() {
return properties;
}
public void setProperties(Map<String,String> properties) {
this.properties = properties;
if (conf == null) {
conf = HBaseConfiguration.create();
}
for (Entry<String, String> entry: getProperties().entrySet()) {
conf.set(entry.getKey(), entry.getValue());
}
}
public synchronized Configuration configuration() {
if (conf == null) {
conf = HBaseConfiguration.create();
}
return conf;
}
// convenience setters
public void setZookeeperQuorum(String value) {
configuration().set("hbase.zookeeper.quorum", value);
}
public void setZookeeperClientPort(int port) {
configuration().setInt("hbase.zookeeper.property.clientPort", port);
}
protected transient HBaseAdmin admin;
public synchronized HBaseAdmin admin() throws IOException {
if (admin == null) {
admin = new HBaseAdmin(configuration());
}
return admin;
}
@Override
public synchronized void stop() {
isRunning = false;
if (admin != null) {
try {
admin.close();
} catch (IOException e) {
logger.warning("problem closing HBaseAdmin " + admin + " - " + e);
}
admin = null;
}
if (conf != null) {
// HConnectionManager.deleteConnection(conf); // XXX?
conf = null;
}
}
protected transient boolean isRunning = false;
@Override
public boolean isRunning() {
return isRunning;
}
@Override
public void start() {
isRunning = true;
}
public synchronized void reset() {
if (admin != null) {
try {
admin.close();
} catch (IOException e) {
logger.warning("problem closing HBaseAdmin " + admin + " - " + e);
}
admin = null;
}
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/HBaseContentDigestHistory.java
-303
View File
@@ -1,303 +0,0 @@
package org.archive.modules.recrawl.hbase;
import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_CONTENT_DIGEST_COUNT;
import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_ORIGINAL_DATE;
import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_ORIGINAL_URL;
import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_WARC_FILENAME;
import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_WARC_FILE_OFFSET;
import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_WARC_RECORD_ID;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
import org.apache.hadoop.hbase.regionserver.NoSuchColumnFamilyException;
import org.apache.hadoop.hbase.util.Bytes;
import org.archive.modules.CrawlURI;
import org.archive.modules.recrawl.AbstractContentDigestHistory;
import org.json.JSONException;
import org.json.JSONObject;
import org.springframework.context.Lifecycle;
import com.google.common.collect.BiMap;
import com.google.common.collect.HashBiMap;
/**
* HBase content digest history store. Must be a toplevel bean in
* crawler-beans.cxml in order to receive {@link Lifecycle} events.
*
* @see AbstractContentDigestHistory
* @author nlevitt
*/
public class HBaseContentDigestHistory extends AbstractContentDigestHistory implements Lifecycle {
private static final Logger logger =
Logger.getLogger(HBaseContentDigestHistory.class.getName());
protected static final byte[] COLUMN_FAMILY = Bytes.toBytes("f");
protected static final byte[] COLUMN = Bytes.toBytes("c");
protected static final BiMap<String,String> JSON_KEYS_MAP = HashBiMap.create();
static {
JSON_KEYS_MAP.put(A_CONTENT_DIGEST_COUNT, "c");
JSON_KEYS_MAP.put(A_ORIGINAL_URL, "u");
JSON_KEYS_MAP.put(A_WARC_RECORD_ID, "i");
JSON_KEYS_MAP.put(A_WARC_FILENAME, "f");
JSON_KEYS_MAP.put(A_WARC_FILE_OFFSET, "o");
JSON_KEYS_MAP.put(A_ORIGINAL_DATE, "d");
}
protected HBaseTable table;
public void setTable(HBaseTable table) {
this.table = table;
}
protected boolean addColumnFamily = false;
public boolean getAddColumnFamily() {
return addColumnFamily;
}
/**
* Add the expected column family
* {@link #COLUMN_FAMILY} to the HBase table if the
* table doesn't already have it.
*/
public void setAddColumnFamily(boolean addColumnFamily) {
this.addColumnFamily = addColumnFamily;
}
protected int retryIntervalMs = 10*1000;
public int getRetryIntervalMs() {
return retryIntervalMs;
}
public void setRetryIntervalMs(int retryIntervalMs) {
this.retryIntervalMs = retryIntervalMs;
}
protected int maxTries = 1;
public int getMaxTries() {
return maxTries;
}
public void setMaxTries(int maxTries) {
this.maxTries = maxTries;
}
protected String keySuffix = null;
public String getKeySuffix() {
return keySuffix;
}
/**
* If not null, keySuffix is appended to the lookup key when loading and
* storing digest history. Thus the key looks like {digest}{keySuffix}, e.g.
* "sha1:22SFHXERHNFOEY6WK7YOUN4PFIPZSB4D-1193". The purpose is to support
* multiple namespaces in a single hbase table, to avoid proliferation of
* small tables. The reason we use a suffix instead of a prefix is to leave
* open the possibility of deduplication across these different namespaces
* at some point in the future.
*
* @param keySuffix
*/
public void setKeySuffix(String keySuffix) {
this.keySuffix = keySuffix;
}
@Override
protected String persistKeyFor(CrawlURI curi) {
if (keySuffix != null) {
return super.persistKeyFor(curi) + keySuffix;
} else {
return super.persistKeyFor(curi);
}
}
protected synchronized void addColumnFamily() {
try {
HTableDescriptor oldDesc = table.getHtableDescriptor();
if (oldDesc.getFamily(COLUMN_FAMILY) == null) {
HTableDescriptor newDesc = new HTableDescriptor(oldDesc);
newDesc.addFamily(new HColumnDescriptor(COLUMN_FAMILY));
logger.info("table does not yet have expected column family, modifying descriptor to " + newDesc);
HBaseAdmin hbaseAdmin = table.getHbase().admin();
hbaseAdmin.disableTable(table.getName());
hbaseAdmin.modifyTable(Bytes.toBytes(table.getName()), newDesc);
hbaseAdmin.enableTable(table.getName());
}
} catch (IOException e) {
logger.warning("problem adding column family: " + e);
}
}
private boolean isRunning;
@Override
public void start() {
// add column family here to avoid disabling table while another
// ToeThread is trying to use it
if (getAddColumnFamily()) {
addColumnFamily();
}
this.isRunning = true;
}
@Override
public void stop() {
this.isRunning = false;
}
@Override
public boolean isRunning() {
return isRunning;
}
@Override
public void load(CrawlURI curi) {
// make this call in all cases so that the value is initialized and
// WARCWriterProcessor knows it should put the info in there
HashMap<String, Object> contentDigestHistory = curi.getContentDigestHistory();
byte[] key = Bytes.toBytes(persistKeyFor(curi));
Result hbaseResult = tryHbaseGet(curi, new Get(key));
if (hbaseResult != null) {
Map<String, Object> loadedHistory = parseHbaseResult(curi, hbaseResult);
if (loadedHistory != null) {
if (logger.isLoggable(Level.FINER)) {
logger.finer("loaded history by digest " + persistKeyFor(curi)
+ " for uri " + curi + " - " + loadedHistory);
}
contentDigestHistory.putAll(loadedHistory);
}
}
}
protected Result tryHbaseGet(CrawlURI curi, Get hbaseGet) {
try {
return table.get(hbaseGet);
} catch (IOException e) {
logger.warning("problem retrieving persist data from hbase, proceeding without, for digest " + persistKeyFor(curi) + " uri " + curi + " - " + e);
return null;
}
}
protected Map<String, Object> parseHbaseResult(CrawlURI curi, Result hbaseResult) {
HashMap<String, Object> loadedHistory = null;
// no data for uri is indicated by empty Result
if (!hbaseResult.isEmpty()) {
byte[] jsonBytes = hbaseResult.getValue(COLUMN_FAMILY, COLUMN);
if (jsonBytes != null) {
JSONObject json = null;
try {
json = new JSONObject(Bytes.toString(jsonBytes));
loadedHistory = new HashMap<String,Object>();
@SuppressWarnings("unchecked")
Iterator<String> keyIter = json.keys();
while (keyIter.hasNext()) {
String jsonKey = keyIter.next();
Object jsonValue = json.get(jsonKey);
String historyMapKey = JSON_KEYS_MAP.inverse().get(jsonKey);
if (historyMapKey == null) {
logger.warning("unknown key \"" + jsonKey + "\" found in hbase json for digest " + persistKeyFor(curi));
historyMapKey = jsonKey;
}
loadedHistory.put(historyMapKey, jsonValue);
}
} catch (JSONException e) {
logger.warning("problem parsing json for digest " + persistKeyFor(curi) + " uri " + curi + " - " + e);
}
} else {
// shouldn't happen? result.isEmpty() is normal case
logger.fine("[jsonBytes==null] no persist data for digest " + persistKeyFor(curi) + " uri " + curi);
}
} else {
logger.finest("[result.isEmpty()] no persist data for digest " + persistKeyFor(curi) + " uri " + curi);
}
return loadedHistory;
}
@Override
public void store(CrawlURI curi) {
if (!curi.hasContentDigestHistory()
|| curi.getContentDigestHistory().isEmpty()) {
return;
}
if (logger.isLoggable(Level.FINER)) {
logger.finer("storing history by digest " + persistKeyFor(curi)
+ " for uri " + curi + " - "
+ curi.getContentDigestHistory());
}
Put hbasePut = createHbasePut(curi);
tryHbasePut(curi, hbasePut);
}
protected void tryHbasePut(CrawlURI curi, Put p) {
int tryCount = 0;
do {
tryCount++;
try {
table.put(p);
return;
} catch (RetriesExhaustedWithDetailsException e) {
if (e.getCause(0) instanceof NoSuchColumnFamilyException && getAddColumnFamily()) {
addColumnFamily();
tryCount--;
} else {
logger.warning("put failed " + "(try " + tryCount + " of "
+ getMaxTries() + ")" + " for " + curi + " - " + e);
}
} catch (IOException e) {
logger.warning("put failed " + "(try " + tryCount + " of "
+ getMaxTries() + ")" + " for " + curi + " - " + e);
} catch (NullPointerException e) {
// HTable.put() throws NullPointerException while connection is lost.
logger.warning("put failed " + "(try " + tryCount + " of "
+ getMaxTries() + ")" + " for " + curi + " - " + e);
}
if (tryCount > 0 && tryCount < getMaxTries() && isRunning()) {
try {
Thread.sleep(getRetryIntervalMs());
} catch (InterruptedException ex) {
logger.warning("thread interrupted. aborting retry for " + curi);
return;
}
}
} while (tryCount < getMaxTries() && isRunning());
if (isRunning()) {
logger.warning("giving up after " + tryCount + " tries on put for " + curi);
}
}
protected Put createHbasePut(CrawlURI curi) {
byte[] key = Bytes.toBytes(persistKeyFor(curi));
Put hbasePut = new Put(key);
try {
JSONObject json = new JSONObject();
for (Entry<String, Object> entry: curi.getContentDigestHistory().entrySet()) {
String jsonKey = JSON_KEYS_MAP.get(entry.getKey());
if (jsonKey == null) {
logger.warning("unknown key \"" + entry.getKey() + "\" found in content digest history map for " + curi);
jsonKey = entry.getKey();
}
json.put(jsonKey, entry.getValue());
}
hbasePut.add(COLUMN_FAMILY, COLUMN, Bytes.toBytes(json.toString()));
} catch (JSONException e) {
// should not happen - all values are either primitive or String.
logger.log(Level.SEVERE, "problem creating json object for digest " + persistKeyFor(curi) + " uri " + curi, e);
}
return hbasePut;
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/HBasePersistLoadProcessor.java
-88
View File
@@ -1,88 +0,0 @@
/*
* This file is part of the Heritrix web crawler (crawler.archive.org).
*
* Licensed to the Internet Archive (IA) by one or more individual
* contributors.
*
* The IA 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.archive.modules.recrawl.hbase;
import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.archive.modules.CrawlURI;
import org.archive.modules.ProcessResult;
import org.archive.modules.Processor;
import org.archive.modules.recrawl.FetchHistoryProcessor;
/**
* A {@link Processor} for retrieving recrawl info from HBase table.
* See {@link HBasePersistProcessor} for table schema.
* As with other fetch history processors, this needs to be combined with {@link FetchHistoryProcessor}
* (set up after FetchHTTP, before WarcWriter) to work.
* @see HBasePersistStoreProcessor
* @author kenji
*/
public class HBasePersistLoadProcessor extends HBasePersistProcessor {
private static final Logger logger =
Logger.getLogger(HBasePersistLoadProcessor.class.getName());
@Override
protected ProcessResult innerProcessResult(CrawlURI uri) throws InterruptedException {
byte[] key = rowKeyForURI(uri);
Get g = new Get(key);
try {
Result r = table.get(g);
// no data for uri is indicated by empty Result
if (r.isEmpty()) {
if (logger.isLoggable(Level.FINE)) {
logger.fine(uri + ": <no crawlinfo>");
}
return ProcessResult.PROCEED;
}
schema.load(r, uri);
if (uri.getFetchStatus() < 0) {
return ProcessResult.FINISH;
}
} catch (IOException e) {
logger.warning("problem retrieving persist data from hbase, proceeding without, for " + uri + " - " + e);
} catch (Exception ex) {
// get() throws RuntimeException upon ZooKeeper connection failures.
// no crawl history load failure should make fetch of URL fail.
logger.log(Level.WARNING, "Get failed for " + uri + ": ", ex);
}
return ProcessResult.PROCEED;
}
/**
* unused.
*/
@Override
protected void innerProcess(CrawlURI uri) throws InterruptedException {
}
@Override
protected boolean shouldProcess(CrawlURI uri) {
// TODO: we want deduplicate robots.txt, too.
//if (uri.isPrerequisite()) return false;
String scheme = uri.getUURI().getScheme();
if (!(scheme.equals("http") || scheme.equals("https") || scheme.equals("ftp") || scheme.equals("sftp"))) {
return false;
}
return true;
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/HBasePersistProcessor.java
-32
View File
@@ -1,32 +0,0 @@
package org.archive.modules.recrawl.hbase;
import org.archive.modules.CrawlURI;
import org.archive.modules.recrawl.AbstractPersistProcessor;
import org.springframework.beans.factory.annotation.Required;
/**
* A base class for processors for keeping de-duplication data in HBase.
* Table schema is defined by {@link RecrawlDataSchema} implementation.
* @author kenji
*/
public abstract class HBasePersistProcessor extends AbstractPersistProcessor {
protected HBaseTableBean table;
@Required
public void setTable(HBaseTableBean table) {
this.table = table;
}
protected RecrawlDataSchema schema;
public RecrawlDataSchema getSchema() {
return schema;
}
@Required
public void setSchema(RecrawlDataSchema schema) {
this.schema = schema;
}
protected byte[] rowKeyForURI(CrawlURI curi) {
return schema.rowKeyForURI(curi);
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/HBasePersistStoreProcessor.java
-133
View File
@@ -1,133 +0,0 @@
/*
* This file is part of the Heritrix web crawler (crawler.archive.org).
*
* Licensed to the Internet Archive (IA) by one or more individual
* contributors.
*
* The IA 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.archive.modules.recrawl.hbase;
import java.io.IOException;
import java.util.logging.Logger;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
import org.apache.hadoop.hbase.regionserver.NoSuchColumnFamilyException;
import org.apache.hadoop.hbase.util.Bytes;
import org.archive.modules.CrawlURI;
import org.archive.modules.fetcher.FetchStatusCodes;
import org.archive.modules.recrawl.RecrawlAttributeConstants;
/**
* @author kenji
*/
public class HBasePersistStoreProcessor extends HBasePersistProcessor implements FetchStatusCodes, RecrawlAttributeConstants {
private static final Logger logger = Logger.getLogger(HBasePersistStoreProcessor.class.getName());
protected boolean addColumnFamily = false;
public boolean getAddColumnFamily() {
return addColumnFamily;
}
/**
* Add the expected column family
* {@link HBaseContentDigestHistory#COLUMN_FAMILY} to the HBase table if the
* table doesn't already have it.
*/
public void setAddColumnFamily(boolean addColumnFamily) {
this.addColumnFamily = addColumnFamily;
}
protected int retryIntervalMs = 10*1000;
public int getRetryIntervalMs() {
return retryIntervalMs;
}
public void setRetryIntervalMs(int retryIntervalMs) {
this.retryIntervalMs = retryIntervalMs;
}
protected int maxTries = 1;
public int getMaxTries() {
return maxTries;
}
public void setMaxTries(int maxTries) {
this.maxTries = maxTries;
}
protected synchronized void addColumnFamily() {
try {
HTableDescriptor oldDesc = table.getHtableDescriptor();
byte[] columnFamily = Bytes.toBytes(schema.getColumnFamily());
if (oldDesc.getFamily(columnFamily) == null) {
HTableDescriptor newDesc = new HTableDescriptor(oldDesc);
newDesc.addFamily(new HColumnDescriptor(columnFamily));
logger.info("table does not yet have expected column family, modifying descriptor to " + newDesc);
HBaseAdmin hbaseAdmin = table.getHbase().admin();
hbaseAdmin.disableTable(table.getName());
hbaseAdmin.modifyTable(Bytes.toBytes(table.getName()), newDesc);
hbaseAdmin.enableTable(table.getName());
}
} catch (IOException e) {
logger.warning("problem adding column family: " + e);
}
}
@Override
protected void innerProcess(CrawlURI uri) {
Put p = schema.createPut(uri);
int tryCount = 0;
do {
tryCount++;
try {
table.put(p);
return;
} catch (RetriesExhaustedWithDetailsException e) {
if (e.getCause(0) instanceof NoSuchColumnFamilyException && getAddColumnFamily()) {
addColumnFamily();
tryCount--;
} else {
logger.warning("put failed " + "(try " + tryCount + " of "
+ getMaxTries() + ")" + " for " + uri + " - " + e);
}
} catch (IOException e) {
logger.warning("put failed " + "(try " + tryCount + " of "
+ getMaxTries() + ")" + " for " + uri + " - " + e);
} catch (NullPointerException e) {
// HTable.put() throws NullPointerException while connection is lost.
logger.warning("put failed " + "(try " + tryCount + " of "
+ getMaxTries() + ")" + " for " + uri + " - " + e);
}
if (tryCount > 0 && tryCount < getMaxTries() && isRunning()) {
try {
Thread.sleep(getRetryIntervalMs());
} catch (InterruptedException ex) {
logger.warning("thread interrupted. aborting retry for " + uri);
return;
}
}
} while (tryCount < getMaxTries() && isRunning());
if (isRunning()) {
logger.warning("giving up after " + tryCount + " tries on put for " + uri);
}
}
@Override
protected boolean shouldProcess(CrawlURI curi) {
return super.shouldStore(curi);
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/HBaseTable.java
-159
View File
@@ -1,159 +0,0 @@
/*
* This file is part of the Heritrix web crawler (crawler.archive.org).
*
* Licensed to the Internet Archive (IA) by one or more individual
* contributors.
*
* The IA 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.archive.modules.recrawl.hbase;
import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
/**
* @author kenji
* @author nlevitt
*/
public class HBaseTable extends HBaseTableBean {
static final Logger logger =
Logger.getLogger(HBaseTable.class.getName());
protected boolean create = false;
protected HConnection hconn = null;
protected ThreadLocal<HTableInterface> htable = new ThreadLocal<HTableInterface>();
public boolean getCreate() {
return create;
}
/** Create the named table if it doesn't exist. */
public void setCreate(boolean create) {
this.create = create;
}
public HBaseTable() {
}
protected synchronized HConnection hconnection() throws IOException {
if (hconn == null) {
hconn = HConnectionManager.createConnection(hbase.configuration());
}
return hconn;
}
protected HTableInterface htable() throws IOException {
if (htable.get() == null) {
htable.set(hconnection().getTable(htableName));
}
return htable.get();
}
@Override
public void put(Put p) throws IOException {
try {
htable().put(p);
} catch (IOException e) {
reset();
throw e;
}
}
@Override
public Result get(Get g) throws IOException {
try {
return htable().get(g);
} catch (IOException e) {
reset();
throw e;
}
}
public HTableDescriptor getHtableDescriptor() throws IOException {
try {
return htable().getTableDescriptor();
} catch (IOException e) {
reset();
throw e;
}
}
@Override
public void start() {
if (getCreate()) {
int attempt = 1;
while (true) {
try {
HBaseAdmin admin = hbase.admin();
if (!admin.tableExists(htableName)) {
HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(htableName));
logger.info("hbase table '" + htableName + "' does not exist, creating it... " + desc);
admin.createTable(desc);
}
break;
} catch (IOException e) {
logger.log(Level.WARNING, "(attempt " + attempt + ") problem creating hbase table " + htableName, e);
attempt++;
reset();
// back off up to 60 seconds between retries
try {
Thread.sleep(Math.min(attempt * 1000, 60000));
} catch (InterruptedException e1) {
}
}
}
}
super.start();
}
protected void reset() {
if (htable.get() != null) {
try {
htable.get().close();
} catch (IOException e) {
logger.log(Level.WARNING, "htablename='" + htableName + "' htable.close() threw " + e, e);
}
htable.remove();
}
if (hconn != null) {
try {
hconn.close();
} catch (IOException e) {
logger.log(Level.WARNING, "hconn.close() threw " + e, e);
}
// HConnectionManager.deleteStaleConnection(hconn);
hconn = null;
}
hbase.reset();
}
@Override
public synchronized void stop() {
super.stop();
reset();
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/HBaseTableBean.java
-78
View File
@@ -1,78 +0,0 @@
package org.archive.modules.recrawl.hbase;
import java.io.IOException;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.archive.modules.recrawl.PersistOnlineProcessor;
import org.springframework.context.Lifecycle;
/**
* base class for different types of HBaseTable Spring bean implementations.
* @author kenji
* @author nlevitt
*
*/
public abstract class HBaseTableBean implements Lifecycle {
protected String htableName = PersistOnlineProcessor.URI_HISTORY_DBNAME;
protected HBase hbase = new HBase();
protected transient boolean isRunning = false;
// <backward-compatibility>
public void setName(String name) {
this.htableName = name;
}
public String getName() {
return htableName;
}
// </backward-compatibility>
/**
* set name of single HTable this instance accesses.
* @param htableName
*/
public void setHtableName(String htableName) {
this.htableName = htableName;
}
public String getHtableName() {
return htableName;
}
public HBaseTableBean() {
super();
}
public void setHbase(HBase hbase) {
this.hbase = hbase;
}
public HBase getHbase() {
return hbase;
}
public abstract void put(Put p) throws IOException;
public abstract Result get(Get g) throws IOException;
public abstract HTableDescriptor getHtableDescriptor() throws IOException;
@Override
public boolean isRunning() {
return isRunning;
}
@Override
public void start() {
isRunning = true;
}
@Override
public synchronized void stop() {
isRunning = false;
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/MultiColumnRecrawlDataSchema.java
-140
View File
@@ -1,140 +0,0 @@
/*
* This file is part of the Heritrix web crawler (crawler.archive.org).
*
* Licensed to the Internet Archive (IA) by one or more individual
* contributors.
*
* The IA 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.archive.modules.recrawl.hbase;
import java.util.Map;
import java.util.logging.Logger;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.archive.modules.CrawlURI;
import org.archive.modules.fetcher.FetchStatusCodes;
import org.archive.modules.recrawl.FetchHistoryHelper;
import org.archive.modules.recrawl.RecrawlAttributeConstants;
/**
* RecrawlDataSchema that stores each recrawl data properties in a separate column in single column
* family, whose name may be configured with {@link #setColumnFamily(String)} (default "f").
* <ul>
* <li>{@code s}: fetch status (as integer text)</li>
* <li>{@code d}: content digest (with {@code sha1:} prefix, Base32 text)</li>
* <li>{@code e}: ETag (enclosing quotes stripped)</li>
* <li>{@code m}: last-modified date-time (as integer timestamp, binary format)</li>
* <li>{@code z}: do-not-crawl flag - loader discards URL if this column has non-empty value.</li>
* </ul>
*
* @author kenji
*/
public class MultiColumnRecrawlDataSchema extends RecrawlDataSchemaBase implements RecrawlDataSchema, RecrawlAttributeConstants {
static final Logger logger = Logger.getLogger(MultiColumnRecrawlDataSchema.class.getName());
public static final byte[] COLUMN_STATUS = Bytes.toBytes("s");
public static final byte[] COLUMN_CONTENT_DIGEST = Bytes.toBytes("d");
public static final byte[] COLUMN_ETAG = Bytes.toBytes("e");
public static final byte[] COLUMN_LAST_MODIFIED = Bytes.toBytes("m");
/* (non-Javadoc)
* @see org.archive.modules.hq.recrawl.RecrawlDataSchema#createPut()
*/
public Put createPut(CrawlURI uri) {
byte[] uriBytes = rowKeyForURI(uri);
byte[] key = uriBytes;
Put p = new Put(key);
String digest = uri.getContentDigestSchemeString();
if (digest != null) {
p.add(columnFamily, COLUMN_CONTENT_DIGEST, Bytes.toBytes(digest));
}
p.add(columnFamily, COLUMN_STATUS, Bytes.toBytes(Integer.toString(uri.getFetchStatus())));
if (uri.isHttpTransaction()) {
String etag = uri.getHttpResponseHeader(RecrawlAttributeConstants.A_ETAG_HEADER);
if (etag != null) {
// Etqg is usually quoted
if (etag.length() >= 2 && etag.charAt(0) == '"' && etag.charAt(etag.length() - 1) == '"')
etag = etag.substring(1, etag.length() - 1);
p.add(columnFamily, COLUMN_ETAG, Bytes.toBytes(etag));
}
String lastmod = uri.getHttpResponseHeader(RecrawlAttributeConstants.A_LAST_MODIFIED_HEADER);
if (lastmod != null) {
long lastmod_sec = FetchHistoryHelper.parseHttpDate(lastmod);
if (lastmod_sec == 0) {
try {
lastmod_sec = uri.getFetchCompletedTime();
} catch (NullPointerException ex) {
logger.warning("CrawlURI.getFetchCompletedTime():" + ex + " for " + uri.shortReportLine());
}
}
if (lastmod_sec != 0)
p.add(columnFamily, COLUMN_LAST_MODIFIED, Bytes.toBytes(lastmod_sec));
} else {
try {
long completed = uri.getFetchCompletedTime();
if (completed != 0)
p.add(columnFamily, COLUMN_LAST_MODIFIED, Bytes.toBytes(completed));
} catch (NullPointerException ex) {
logger.warning("CrawlURI.getFetchCompletedTime():" + ex + " for " + uri.shortReportLine());
}
}
}
return p;
}
/* (non-Javadoc)
* @see org.archive.modules.hq.recrawl.RecrawlDataSchema#load(java.util.Map, org.apache.hadoop.hbase.client.Result)
*/
public void load(Result result, CrawlURI curi) {
// check for "do-not-crawl" flag - any non-empty data tells not to crawl this
// URL.
byte[] nocrawl = result.getValue(columnFamily, COLUMN_NOCRAWL);
if (nocrawl != null && nocrawl.length > 0) {
// fetch status set to S_DEEMED_CHAFF, because this do-not-crawl flag
// is primarily intended for preventing crawler from stepping on traps.
curi.setFetchStatus(FetchStatusCodes.S_DEEMED_CHAFF);
curi.getAnnotations().add("nocrawl");
return;
}
// all column should have identical timestamp.
KeyValue rkv = result.getColumnLatest(columnFamily, COLUMN_STATUS);
long timestamp = rkv.getTimestamp();
Map<String, Object> history = FetchHistoryHelper.getFetchHistory(curi, timestamp, historyLength);
// FetchHTTP ignores history with status <= 0
byte[] status = result.getValue(columnFamily, COLUMN_STATUS);
if (status != null) {
// Note that status is stored as integer text. It's typically three-chars
// that is less than 4-byte integer bits.
history.put(RecrawlAttributeConstants.A_STATUS, Integer.parseInt(Bytes.toString(status)));
byte[] etag = result.getValue(columnFamily, COLUMN_ETAG);
if (etag != null) {
history.put(RecrawlAttributeConstants.A_ETAG_HEADER, Bytes.toString(etag));
}
byte[] lastmod = result.getValue(columnFamily, COLUMN_LAST_MODIFIED);
if (lastmod != null) {
long lastmod_sec = Bytes.toLong(lastmod);
history.put(RecrawlAttributeConstants.A_LAST_MODIFIED_HEADER, FetchHistoryHelper.formatHttpDate(lastmod_sec));
}
byte[] digest = result.getValue(columnFamily, COLUMN_CONTENT_DIGEST);
if (digest != null) {
history.put(RecrawlAttributeConstants.A_CONTENT_DIGEST, Bytes.toString(digest));
}
}
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/RecrawlDataSchema.java
-35
View File
@@ -1,35 +0,0 @@
/*
* This file is part of the Heritrix web crawler (crawler.archive.org).
*
* Licensed to the Internet Archive (IA) by one or more individual
* contributors.
*
* The IA 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.archive.modules.recrawl.hbase;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.archive.modules.CrawlURI;
/**
* @author kenji
*/
public interface RecrawlDataSchema {
public String getColumnFamily();
public Put createPut(CrawlURI uri);
public void load(Result result, CrawlURI curi);
// TODO: drop this method by revising createPut(CrawlURI) method.
public byte[] rowKeyForURI(CrawlURI curi);
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/RecrawlDataSchemaBase.java
-123
View File
@@ -1,123 +0,0 @@
package org.archive.modules.recrawl.hbase;
import java.util.Map;
import java.util.logging.Logger;
import org.apache.hadoop.hbase.util.Bytes;
import org.archive.modules.CrawlURI;
import org.archive.modules.canonicalize.CanonicalizationRule;
import org.archive.modules.recrawl.FetchHistoryHelper;
import org.archive.modules.recrawl.FetchHistoryProcessor;
import org.archive.modules.recrawl.PersistProcessor;
/**
* implements common utility methods for implementing {@link RecrawlDataSchema}.
* <ul>
* <li>configuring single column family name</li>
* <li>formatting/parsing HTTP date text</li>
* <li>constructing row key</li>
* <li>preparing fetch-history array</li>
* </ul>
* @author kenji
*/
abstract public class RecrawlDataSchemaBase implements RecrawlDataSchema {
private static final Logger logger = Logger.getLogger(RecrawlDataSchemaBase.class.getName());
/**
* default value for {@link #columnFamily}.
*/
public static final byte[] DEFAULT_COLUMN_FAMILY = Bytes.toBytes("f");
protected byte[] columnFamily = DEFAULT_COLUMN_FAMILY;
public static final byte[] COLUMN_NOCRAWL = Bytes.toBytes("z");
/**
* default value for {@link #useCanonicalString}.
*/
public static boolean DEFAULT_USE_CANONICAL_STRING = true;
private boolean useCanonicalString = DEFAULT_USE_CANONICAL_STRING;
private CanonicalizationRule keyRule = null;
protected int historyLength = 2;
public RecrawlDataSchemaBase() {
super();
}
public void setColumnFamily(String colf) {
columnFamily = Bytes.toBytes(colf);
}
public boolean isUseCanonicalString() {
return useCanonicalString;
}
/**
* if set to true, canonicalized string will be used as row key, rather than URI
* @param useCanonicalString
*/
public void setUseCanonicalString(boolean useCanonicalString) {
this.useCanonicalString = useCanonicalString;
}
public String getColumnFamily() {
return Bytes.toString(columnFamily);
}
public CanonicalizationRule getKeyRule() {
return keyRule;
}
/**
* alternative canonicalization rule for generating row key from URI.
* TODO: currently unused.
* @param keyRule
*/
public void setKeyRule(CanonicalizationRule keyRule) {
this.keyRule = keyRule;
}
public int getHistoryLength() {
return historyLength;
}
/**
* maximum number of crawl history entries to retain in {@link CrawlURI}.
* when more than this number of crawl history entry is being added by
* {@link #getFetchHistory(CrawlURI, long)}, oldest entry will be discarded.
* {@code historyLength} should be the same number as
* {@link FetchHistoryProcessor#setHistoryLength(int)}, or FetchHistoryProcessor will
* reallocate the crawl history array.
* @param historyLength
* @see FetchHistoryProcessor#setHistoryLength(int)
*/
public void setHistoryLength(int historyLength) {
this.historyLength = historyLength;
}
/**
* calls {@link FetchHistoryHelper#getFetchHistory(CrawlURI, long, int)} with {@link #historyLength}.
* @param uri CrawlURI from which fetch history is obtained.
* @return Map object for storing re-crawl data (never null).
* @see FetchHistoryHelper#getFetchHistory(CrawlURI, long, int)
* @see FetchHistoryProcessor
*/
protected Map<String, Object> getFetchHistory(CrawlURI uri, long timestamp) {
return FetchHistoryHelper.getFetchHistory(uri, timestamp, historyLength);
}
/**
* return row key for {@code curi}.
* TODO: move this to HBasePersistProcessor by redesigning {@link RecrawlDataSchema}.
* @param curi {@link CrawlURI} for which a row is being fetched.
* @return row key
*/
public byte[] rowKeyForURI(CrawlURI curi) {
if (useCanonicalString) {
// TODO: use keyRule if specified.
return Bytes.toBytes(PersistProcessor.persistKeyFor(curi));
} else {
return Bytes.toBytes(curi.toString());
}
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/SingleColumnJsonRecrawlDataSchema.java
-174
View File
@@ -1,174 +0,0 @@
/*
* This file is part of the Heritrix web crawler (crawler.archive.org).
*
* Licensed to the Internet Archive (IA) by one or more individual
* contributors.
*
* The IA 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.archive.modules.recrawl.hbase;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.commons.httpclient.HttpMethod;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.archive.modules.CrawlURI;
import org.archive.modules.fetcher.FetchStatusCodes;
import org.archive.modules.recrawl.FetchHistoryHelper;
import org.archive.modules.recrawl.RecrawlAttributeConstants;
import org.json.JSONException;
import org.json.JSONObject;
/**
* {@linkplain SingleColumnJsonRecrawlDataSchema} stores all re-crawl data properties in a single column,
* in JSON format. As HBase stores each column paired with the row key, it takes a lot of space to store
* each re-crawl data property in its own column.
* <ul>
* <li>{@code r}: re-crawl data in JSON format</li>
* <li>{@code z}: do-not-crawl flag - loader discards URL if this column has non-empty value.</li>
* </ul>
* @author Kenji Nagahashi
*/
public class SingleColumnJsonRecrawlDataSchema extends RecrawlDataSchemaBase
implements RecrawlDataSchema {
static final Logger logger = Logger.getLogger(SingleColumnJsonRecrawlDataSchema.class.getName());
public static byte[] DEFAULT_COLUMN = Bytes.toBytes("r");
// JSON property names for re-crawl data properties
public static final String PROPERTY_STATUS = "s";
public static final String PROPERTY_CONTENT_DIGEST = "d";
public static final String PROPERTY_ETAG = "e";
public static final String PROPERTY_LAST_MODIFIED = "m";
// SHA1 scheme is assumed.
public static final String CONTENT_DIGEST_SCHEME = "sha1:";
// single column for storing JSON of re-crawl data
protected byte[] column = DEFAULT_COLUMN;
public void setColumn(String column) {
this.column = Bytes.toBytes(column);
}
public String getColumn() {
return Bytes.toString(column);
}
/* (non-Javadoc)
* @see org.archive.modules.hq.recrawl.RecrawlDataSchema#createPut(org.archive.modules.CrawlURI)
*/
public Put createPut(CrawlURI uri) {
byte[] key = rowKeyForURI(uri);
Put p = new Put(key);
JSONObject jo = new JSONObject();
try {
// TODO should we post warning message when scheme != "sha1"?
String digest = uri.getContentDigestString();
if (digest != null) {
jo.put(PROPERTY_CONTENT_DIGEST, digest);
}
jo.put(PROPERTY_STATUS, uri.getFetchStatus());
if (uri.isHttpTransaction()) {
String etag = uri.getHttpResponseHeader(RecrawlAttributeConstants.A_ETAG_HEADER);
if (etag != null) {
// Etag is usually quoted
if (etag.length() >= 2 && etag.charAt(0) == '"' && etag.charAt(etag.length() - 1) == '"')
etag = etag.substring(1, etag.length() - 1);
jo.put(PROPERTY_ETAG, etag);
}
String lastmod = uri.getHttpResponseHeader(RecrawlAttributeConstants.A_LAST_MODIFIED_HEADER);
if (lastmod != null) {
long lastmod_sec = FetchHistoryHelper.parseHttpDate(lastmod);
if (lastmod_sec == 0) {
try {
lastmod_sec = uri.getFetchCompletedTime();
} catch (NullPointerException ex) {
logger.warning("CrawlURI.getFetchCompletedTime():" + ex + " for " + uri.shortReportLine());
}
}
} else {
try {
long completed = uri.getFetchCompletedTime();
if (completed != 0)
jo.put(PROPERTY_LAST_MODIFIED, completed);
} catch (NullPointerException ex) {
logger.warning("CrawlURI.getFetchCompletedTime():" + ex + " for " + uri.shortReportLine());
}
}
}
} catch (JSONException ex) {
// should not happen - all values are either primitive or String.
logger.log(Level.SEVERE, "JSON translation failed", ex);
}
p.add(columnFamily, column, Bytes.toBytes(jo.toString()));
return p;
}
/* (non-Javadoc)
* @see org.archive.modules.hq.recrawl.RecrawlDataSchema#load(org.apache.hadoop.hbase.client.Result)
*/
public void load(Result result, CrawlURI curi) {
// check for "do-not-crawl" flag - any non-empty data tells not to crawl this
// URL.
byte[] nocrawl = result.getValue(columnFamily, COLUMN_NOCRAWL);
if (nocrawl != null && nocrawl.length > 0) {
// fetch status set to S_DEEMED_CHAFF, because this do-not-crawl flag
// is primarily intended for preventing crawler from stepping on traps.
curi.setFetchStatus(FetchStatusCodes.S_DEEMED_CHAFF);
curi.getAnnotations().add("nocrawl");
return;
}
KeyValue rkv = result.getColumnLatest(columnFamily, column);
long timestamp = rkv.getTimestamp();
Map<String, Object> history = FetchHistoryHelper.getFetchHistory(curi, timestamp, historyLength);
if (history == null) {
// crawl history array is fully occupied by crawl history entries
// newer than timestamp.
return;
}
byte[] jsonBytes = rkv.getValue();
if (jsonBytes != null) {
JSONObject jo = null;
try {
jo = new JSONObject(Bytes.toString(jsonBytes));
} catch (JSONException ex) {
logger.warning(String.format("JSON parsing failed for key %1s: %2s",
result.getRow(), ex.getMessage()));
}
if (jo != null) {
int status = jo.optInt(PROPERTY_STATUS, -1);
if (status >= 0) {
history.put(RecrawlAttributeConstants.A_STATUS, status);
}
String digest = jo.optString(PROPERTY_CONTENT_DIGEST);
if (digest != null) {
history.put(RecrawlAttributeConstants.A_CONTENT_DIGEST, CONTENT_DIGEST_SCHEME + digest);
}
String etag = jo.optString(PROPERTY_ETAG);
if (etag != null) {
history.put(RecrawlAttributeConstants.A_ETAG_HEADER, etag);
}
long lastmod = jo.optLong(PROPERTY_LAST_MODIFIED);
if (lastmod > 0) {
history.put(RecrawlAttributeConstants.A_LAST_MODIFIED_HEADER, FetchHistoryHelper.formatHttpDate(lastmod));
}
}
}
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/SingleHBaseTable.java
-380
View File
@@ -1,380 +0,0 @@
/*
* This file is part of the Heritrix web crawler (crawler.archive.org).
*
* Licensed to the Internet Archive (IA) by one or more individual
* contributors.
*
* The IA 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.archive.modules.recrawl.hbase;
import java.io.IOException;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.NotServingRegionException;
import org.apache.hadoop.hbase.TableNotFoundException;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
/**
* simple HTable wrapper that shares single instance of HTable among threads.
* If you only perform get on HTable, this implementation
* should be good enough. If multiple threads performs Put, {@link HBaseTable}
* would be more efficient.
* <p>when HBase I/O fails due to issue with network/region server/zookeeper, this
* class waits for preset time (see {@link #setReconnectInterval(int)})
* before trying to reestablish HBase connection. During this hold-ff period, all
* {@link #get(Get)} and {@link #put(Put)} calls will fail.
*
* @author kenji
*/
public class SingleHBaseTable extends HBaseTableBean {
private static final Log LOG = LogFactory.getLog(SingleHBaseTable.class);
private HTableInterface table;
private volatile long tableError;
private ReentrantReadWriteLock tableUseLock = new ReentrantReadWriteLock();
boolean autoReconnect = true;
public boolean isAutoReconnect() {
return autoReconnect;
}
/**
* if set to {@code true}, HBaseClient tries to reconnect to the HBase master
* immediately when Put request failed due to connection loss (note {@link #put(Put)}
* still throws IOException even if autoReconnect is enabled.)
* @param autoReconnect true to enable auto-reconnect
*/
public void setAutoReconnect(boolean autoReconnect) {
this.autoReconnect = autoReconnect;
}
protected boolean autoFlush = true;
/**
* passed on to HTable's autoFlush property upon creation.
* @return true for enabling auto-flush.
*/
public boolean isAutoFlush() {
return autoFlush;
}
public void setAutoFlush(boolean autoFlush) {
this.autoFlush = autoFlush;
}
// default 3 minutes
private int reconnectInterval = 1000 * 3 * 60;
public int getReconnectInterval() {
return reconnectInterval;
}
/**
* set hold-off interval upon communication errors.
* @param reconnectInterval hold-off interval in milliseconds.
*/
public void setReconnectInterval(int reconnectInterval) {
this.reconnectInterval = reconnectInterval;
}
// counters
protected AtomicLong getCount = new AtomicLong();
// count of GET/PUT failures (i.e. not counting connection failures).
protected AtomicLong getErrorCount = new AtomicLong();
protected AtomicLong getSkipCount = new AtomicLong();
protected AtomicLong putCount = new AtomicLong();
protected AtomicLong putErrorCount = new AtomicLong();
protected AtomicLong putSkipCount = new AtomicLong();
protected AtomicLong connectCount = new AtomicLong();
public long getGetCount() { return getCount.get(); }
public long getGetErrorCount() { return getErrorCount.get(); }
public long getGetSkipCount() { return getSkipCount.get(); }
public long getPutCount() { return putCount.get(); }
public long getConnectCount() { return connectCount.get(); }
// for diagnosing deadlock situation
public Map<String, Object> getTableLockState() {
Map<String, Object> m = new LinkedHashMap<String, Object>();
m.put("readLockCount", tableUseLock.getReadLockCount());
m.put("queueLength", tableUseLock.getQueueLength());
m.put("writeLocked", tableUseLock.isWriteLocked());
return m;
}
public SingleHBaseTable() {
}
/**
* attempts to reconnect to HBase if table is null.
* must not be called with read-lock.
* @return existing or newly opened HTableInterface.
*/
protected HTableInterface getTable() {
if (table == null && autoReconnect)
openTable();
return table;
}
/**
* close HTable {@code table}, set current time to tableError if closing because
* of a communication error. should be called with write lock.
* @param htable HTable to close.
* @param byError true if closing because of an error.
*/
protected void closeTable(HTableInterface htable, boolean byError) {
if (htable == null) return;
if (table != htable) {
// other thread did closeTable on htable. don't close table.
return;
}
try {
table = null;
htable.close();
} catch (IOException ex) {
LOG.warn("error closing " + htable + " - some commits may have been lost");
}
if (byError) {
tableError = System.currentTimeMillis();
}
}
public void put(Put p) throws IOException {
putCount.incrementAndGet();
// trigger reconnection if necessary. as table can be modified before
// read lock is acquired, we don't read table variable here.
getTable();
boolean htableFailed = false;
HTableInterface htable = null;
Lock readLock = tableUseLock.readLock();
try {
if (!readLock.tryLock(TRY_READ_LOCK_TIMEOUT, TimeUnit.SECONDS)) {
putSkipCount.incrementAndGet();
throw new IOException("could not acquire read lock for HTable.");
}
} catch (InterruptedException ex) {
throw new IOException("interrupted while acquiring read lock", ex);
}
try {
htable = table;
if (htable == null) {
putSkipCount.incrementAndGet();
throw new IOException("HBase connection is unvailable.");
}
// HTable.put() buffers Puts and access to the buffer is not
// synchronized.
synchronized (htable) {
try {
htable.put(p);
} catch (NullPointerException ex) {
// HTable.put() throws NullPointerException when connection is lost.
// It is somewhat weird, so translate it to IOException.
putErrorCount.incrementAndGet();
htableFailed = true;
throw new IOException("hbase connection is lost", ex);
} catch (NotServingRegionException ex) {
putErrorCount.incrementAndGet();
// no need to close HTable.
throw ex;
} catch (IOException ex) {
putErrorCount.incrementAndGet();
htableFailed = true;
throw ex;
}
}
} finally {
readLock.unlock();
if (htableFailed) {
closeTable(htable, true);
}
}
}
public Result get(Get g) throws IOException {
getCount.incrementAndGet();
// trigger reconnection if necessary. as table can be modified before
// read lock is acquired, we don't read table variable here.
getTable();
boolean htableFailed = false;
HTableInterface htable = null;
Lock readLock = tableUseLock.readLock();
try {
if (!readLock.tryLock(TRY_READ_LOCK_TIMEOUT, TimeUnit.SECONDS)) {
getSkipCount.incrementAndGet();
throw new IOException("could not acquire read lock for HTable.");
}
} catch (InterruptedException ex) {
throw new IOException("interrupted while acquiring read lock", ex);
}
try {
htable = table;
if (htable == null) {
getSkipCount.incrementAndGet();
throw new IOException("HBase connection is unvailable.");
}
try {
return htable.get(g);
} catch (NotServingRegionException ex) {
// caused by disruption to HBase cluster. no need to
// refresh HBase connection, since connection itself
// is working okay.
// TODO: should we need to back-off for a while? other
// regions may still be accessible.
getErrorCount.incrementAndGet();
throw ex;
} catch (IOException ex) {
getErrorCount.incrementAndGet();
htableFailed = true;
throw ex;
}
} finally {
readLock.unlock();
if (htableFailed) {
closeTable(htable, true);
}
}
}
@Override
public HTableDescriptor getHtableDescriptor() throws IOException {
HTableInterface table = getTable();
if (table == null) {
throw new IOException("HBase connection is unavailable.");
}
return table.getTableDescriptor();
}
public boolean inBackoffPeriod() {
return (tableError > 0 &&
(System.currentTimeMillis() - tableError) < reconnectInterval);
}
/**
* timestamp of the last Put/Get error.
* @return timestamp in ms.
*/
public long getTableErrorTime() {
return tableError;
}
/**
* connect to HBase.
* it does nothing if table is non-null, or it is in the back-off period since
* the last error.
* should be called with write lock.
*/
protected boolean openTable() {
if (table != null) return true;
// fail immediately if we're in back-off period.
if (inBackoffPeriod()) return false;
try {
HTable t = new HTable(hbase.configuration(), Bytes.toBytes(htableName));
connectCount.incrementAndGet();
t.setAutoFlush(autoFlush);
table = t;
tableError = 0;
return true;
} catch (TableNotFoundException ex) {
// ex.getMessage() only has table name. be a little bit more friendly.
LOG.warn("failed to connect to HTable \"" + htableName + "\": Table Not Found");
tableError = System.currentTimeMillis();
return false;
} catch (IOException ex) {
LOG.warn("failed to connect to HTable \"" + htableName + "\" (" + ex.getMessage() + ")");
tableError = System.currentTimeMillis();
return false;
}
}
/**
* number of seconds to wait for acquiring read lock.
* if read lock is not acquired within this many seconds (probably
* due to deadlock situation on write-lock side), {@link #get(Get)} will
* <i>silently</i> fail.
*/
public final static long TRY_READ_LOCK_TIMEOUT = 5;
/**
* number of seconds to wait for acquiring write lock.
*/
public final static long TRY_WRITE_LOCK_TIMEOUT = 10;
/**
* close current connection and establish new connection.
* fails silently if back-off period is in effect.
*/
protected void reconnect(boolean onerror) throws IOException, InterruptedException {
// avoid deadlock situation caused by attempting
// to acquire write lock while holding read lock.
// there'd be no real dead-lock now that timeout on write lock is implemented,
// but it's nice to know there's a bug in locking.
if (tableUseLock.getReadHoldCount() > 0) {
LOG.warn("avoiding deadlock: reconnect() called by thread with read lock.");
return;
}
Lock writeLock = tableUseLock.writeLock();
if (!writeLock.tryLock(TRY_WRITE_LOCK_TIMEOUT, TimeUnit.SECONDS)) {
LOG.warn("reconnect() could not acquire write lock on tableUseLock for " +
TRY_WRITE_LOCK_TIMEOUT + "s, giving up.");
return;
}
try {
closeTable(table, onerror);
openTable();
} finally {
writeLock.unlock();
}
}
/**
* close current connection and establish new connection.
* for refreshing stale connection through scripting.
* resets tableErrorTime to zero (it will be set to non-zero if
* reconnection attempt fails).
* @throws IOException
* @throws InterruptedException
*/
public void reconnect() throws IOException, InterruptedException {
tableError = 0;
reconnect(false);
}
// public boolean isRunning() {
// return table != null;
// }
public void start() {
super.start();
openTable();
}
public void stop() {
if (table != null) {
try {
table.close();
} catch (IOException ex) {
LOG.warn("table.close() failed", ex);
}
}
table = null;
super.stop();
}
}