trough dedup!

contrib/pom.xml

@@ -61,6 +61,16 @@
 			<artifactId>kafka_2.10</artifactId>
 			<version>0.9.0.0</version>
 		</dependency>
+		<dependency>
+			<groupId>com.rethinkdb</groupId>
+			<artifactId>rethinkdb-driver</artifactId>
+			<version>2.3.3</version>
+		</dependency>
+		<dependency>
+			<groupId>com.googlecode.json-simple</groupId>
+			<artifactId>json-simple</artifactId>
+			<version>1.1.1</version>
+		</dependency>
 	</dependencies>
 	<repositories>
 		<repository>
@@ -105,6 +115,17 @@
 					</execution>
 				</executions>
 			</plugin>
+
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-compiler-plugin</artifactId>
+				<version>3.8.0</version>
+				<configuration>
+					<source>1.8</source>
+					<target>1.8</target>
+					<encoding>UTF-8</encoding>
+				</configuration>
+			</plugin>
 		</plugins>
 	</build>
 </project>

contrib/src/main/java/org/archive/modules/postprocessor/TroughCrawlLogFeed.java

@@ -27,7 +27,6 @@ import java.util.logging.Logger;
 
 import org.apache.commons.collections.Closure;
 import org.apache.commons.io.IOUtils;
-import org.apache.commons.lang.StringEscapeUtils;
 import org.apache.commons.lang.StringUtils;
 import org.apache.http.HttpResponse;
 import org.apache.http.client.methods.HttpPost;
@@ -40,7 +39,7 @@ import org.archive.crawler.frontier.BdbFrontier;
 import org.archive.modules.CrawlURI;
 import org.archive.modules.Processor;
 import org.archive.modules.net.ServerCache;
-import org.archive.util.ArchiveUtils;
+import org.archive.trough.TroughClient;
 import org.archive.util.MimetypeUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.Lifecycle;
@@ -171,18 +170,6 @@ public class TroughCrawlLogFeed extends Processor implements Lifecycle {
         super.stop();
     }
 
-    protected String sqlValue(Object o) {
-        if (o == null) {
-            return "null";
-        } else if (o instanceof Date) {
-            return "datetime('" + ArchiveUtils.getLog14Date((Date) o) + "')";
-        } else if (o instanceof Number) {
-            return o.toString();
-        } else {
-            return "'" + StringEscapeUtils.escapeSql(o.toString()) + "'";
-        }
-    }
-
     transient protected CloseableHttpClient _httpClient;
     protected CloseableHttpClient httpClient() {
         if (_httpClient == null) {
@@ -223,22 +210,22 @@ public class TroughCrawlLogFeed extends Processor implements Lifecycle {
                 warcContentBytes = 0;
             }
             synchronized (crawledBatch) {
-                crawledBatch.add("(" + sqlValue(new Date(curi.getFetchBeginTime())) + ", "
-                        + sqlValue(curi.getFetchStatus()) + ", "
-                        + sqlValue(curi.getContentSize()) + ", "
-                        + sqlValue(curi.getContentLength()) + ", "
-                        + sqlValue(curi) + ", "
-                        + sqlValue(curi.getPathFromSeed()) + ", "
-                        + sqlValue(curi.isSeed() && !"".equals(curi.getPathFromSeed()) ? 1 : 0) + ", "
-                        + sqlValue(curi.getVia()) + ", "
-                        + sqlValue(MimetypeUtils.truncate(curi.getContentType())) + ", "
-                        + sqlValue(curi.getContentDigestSchemeString()) + ", "
-                        + sqlValue(curi.getSourceTag()) + ", "
-                        + sqlValue(curi.isRevisit() ? 1 : 0) + ", "
-                        + sqlValue(curi.getExtraInfo().opt("warcFilename")) + ", "
-                        + sqlValue(curi.getExtraInfo().opt("warcOffset")) + ", "
-                        + sqlValue(warcContentBytes) + ", "
-                        + sqlValue(serverCache.getHostFor(curi.getUURI()).getHostName()) + ")");
+                crawledBatch.add("(" + TroughClient.sqlValue(new Date(curi.getFetchBeginTime())) + ", "
+                        + TroughClient.sqlValue((Object) curi.getFetchStatus()) + ", "
+                        + TroughClient.sqlValue((Object) curi.getContentSize()) + ", "
+                        + TroughClient.sqlValue((Object) curi.getContentLength()) + ", "
+                        + TroughClient.sqlValue(curi) + ", "
+                        + TroughClient.sqlValue(curi.getPathFromSeed()) + ", "
+                        + TroughClient.sqlValue((Object) (curi.isSeed() && !"".equals(curi.getPathFromSeed()) ? 1 : 0)) + ", "
+                        + TroughClient.sqlValue(curi.getVia()) + ", "
+                        + TroughClient.sqlValue(MimetypeUtils.truncate(curi.getContentType())) + ", "
+                        + TroughClient.sqlValue(curi.getContentDigestSchemeString()) + ", "
+                        + TroughClient.sqlValue(curi.getSourceTag()) + ", "
+                        + TroughClient.sqlValue((Object) (curi.isRevisit() ? 1 : 0)) + ", "
+                        + TroughClient.sqlValue(curi.getExtraInfo().opt("warcFilename")) + ", "
+                        + TroughClient.sqlValue(curi.getExtraInfo().opt("warcOffset")) + ", "
+                        + TroughClient.sqlValue((Object) warcContentBytes) + ", "
+                        + TroughClient.sqlValue(serverCache.getHostFor(curi.getUURI()).getHostName()) + ")");
                 if (crawledBatch.size() >= BATCH_MAX_SIZE || System.currentTimeMillis() - crawledBatchLastTime > BATCH_MAX_TIME_MS) {
                     postCrawledBatch();
                 }
@@ -246,13 +233,13 @@ public class TroughCrawlLogFeed extends Processor implements Lifecycle {
         } else {
             synchronized (uncrawledBatch) {
                 uncrawledBatch.add("("
-                        + sqlValue(new Date()) + ", "
-                        + sqlValue(curi) + ", "
-                        + sqlValue(curi.getPathFromSeed()) + ", "
-                        + sqlValue(curi.getFetchStatus()) + ", "
-                        + sqlValue(curi.getVia()) + ", "
-                        + sqlValue(curi.getSourceTag()) + ", "
-                        + sqlValue(serverCache.getHostFor(curi.getUURI()).getHostName()) + ")");
+                        + TroughClient.sqlValue(new Date()) + ", "
+                        + TroughClient.sqlValue(curi) + ", "
+                        + TroughClient.sqlValue(curi.getPathFromSeed()) + ", "
+                        + TroughClient.sqlValue((Object) curi.getFetchStatus()) + ", "
+                        + TroughClient.sqlValue(curi.getVia()) + ", "
+                        + TroughClient.sqlValue(curi.getSourceTag()) + ", "
+                        + TroughClient.sqlValue(serverCache.getHostFor(curi.getUURI()).getHostName()) + ")");
                 if (uncrawledBatch.size() >= BATCH_MAX_SIZE || System.currentTimeMillis() - uncrawledBatchLastTime > BATCH_MAX_TIME_MS) {
                     postUncrawledBatch();
                 }

contrib/src/main/java/org/archive/modules/recrawl/TroughContentDigestHistory.java

@@ -0,0 +1,166 @@
+package org.archive.modules.recrawl;
+
+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_RECORD_ID;
+
+import java.net.MalformedURLException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.TimeZone;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.archive.modules.CrawlURI;
+import org.archive.modules.writer.WARCWriterProcessor;
+import org.archive.spring.HasKeyedProperties;
+import org.archive.spring.KeyedProperties;
+import org.archive.trough.TroughClient;
+import org.archive.util.ArchiveUtils;
+import org.springframework.context.Lifecycle;
+
+/**
+ * AbstractContentDigestHistory implementation for trough.
+ * 
+ * <p>To use, define a {@code TroughContentDigestHistory} top-level bean in your
+ * crawler-beans.cxml, then add {@link ContentDigestHistoryLoader} and
+ * {@link ContentDigestHistoryStorer} to your fetch chain, sandwiching the
+ * {@link WARCWriterProcessor}. In other words, follow the directions at
+ * <a href="https://github.com/internetarchive/heritrix3/wiki/Duplication%20Reduction%20Processors">https://github.com/internetarchive/heritrix3/wiki/Duplication%20Reduction%20Processors</a>
+ * but replace the {@link BdbContentDigestHistory} bean with a
+ * {@code TroughContentDigestHistory} bean.
+ * 
+ * <p>To understand how to use trough as a client, see:
+ * <ul>
+ * <li> <a href="https://github.com/internetarchive/trough/wiki/Some-Notes-About-Trough">https://github.com/internetarchive/trough/wiki/Some-Notes-About-Trough</a>
+ * <li> <a href="https://github.com/internetarchive/trough/blob/repl/trough/client.py">https://github.com/internetarchive/trough/blob/repl/trough/client.py</a>
+ * </ul>
+ * 
+ * @see <a href="https://github.com/internetarchive/warcprox/blob/c70bf2e2b93/warcprox/dedup.py#L480">trough dedup implementation in warcprox</a>
+ */
+public class TroughContentDigestHistory extends AbstractContentDigestHistory implements HasKeyedProperties, Lifecycle {
+    private static final Logger logger = Logger.getLogger(TroughContentDigestHistory.class.getName());
+
+    protected KeyedProperties kp = new KeyedProperties();
+    public KeyedProperties getKeyedProperties() {
+        return kp;
+    }
+
+    public void setSegmentId(String segmentId) {
+        kp.put("segmentId", segmentId);
+    }
+    public String getSegmentId() {
+        return (String) kp.get("segmentId");
+    }
+
+    /**
+     * @param rethinkUrl url with schema rethinkdb:// pointing to
+     *          trough configuration database
+     */
+    public void setRethinkUrl(String rethinkUrl) {
+        kp.put("rethinkUrl", rethinkUrl);
+    }
+    public String getRethinkUrl() {
+        return (String) kp.get("rethinkUrl");
+    }
+
+    protected TroughClient _troughClient = null;
+    protected boolean started = false;
+
+    protected TroughClient troughClient() throws MalformedURLException {
+        if (_troughClient == null) {
+            _troughClient = new TroughClient(getRethinkUrl(), 60 * 60);
+        }
+        return _troughClient;
+    }
+
+    protected static final String SCHEMA_ID = "warcprox-dedup-v1";
+    protected static final String SCHEMA_SQL = "create table dedup (\n"
+            + "    digest_key varchar(100) primary key,\n"
+            + "    url varchar(2100) not null,\n"
+            + "    date datetime not null,\n"
+            + "    id varchar(100));\n"; // warc record id
+
+    @Override
+    public void start() {
+        try {
+            troughClient().registerSchema(SCHEMA_ID, SCHEMA_SQL);
+        } catch (Exception e) {
+            // can happen. hopefully someone else has registered it
+            logger.log(Level.SEVERE, "will try to continue after problem registering schema " + SCHEMA_ID, e);
+        }
+        started  = true;
+    }
+
+    @Override
+    public void stop() {
+    }
+
+    @Override
+    public boolean isRunning() {
+        return started;
+    }
+
+    // dates come back from sqlite in this format: 2019-03-14 00:49:14
+    protected static ThreadLocal<SimpleDateFormat> SQLITE_DATE_FORMAT = new ThreadLocal<SimpleDateFormat>() {
+        protected SimpleDateFormat initialValue() {
+            SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss", Locale.ENGLISH);
+            df.setTimeZone(TimeZone.getTimeZone("GMT"));
+            return df;
+        }
+    };
+
+    @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();
+
+        try {
+            String sql = "select * from dedup where digest_key = %s";
+            List<Map<String, Object>> results = troughClient().read(getSegmentId(), sql, new String[] {persistKeyFor(curi)});
+            if (!results.isEmpty()) {
+                Map<String,Object> hist = new HashMap<String, Object>();
+                hist.put(A_ORIGINAL_URL, results.get(0).get("url"));
+                Date date = SQLITE_DATE_FORMAT.get().parse((String) results.get(0).get("date"));
+                hist.put(A_ORIGINAL_DATE, ArchiveUtils.getLog14Date(date));
+                hist.put(A_WARC_RECORD_ID, results.get(0).get("id"));
+
+                if (logger.isLoggable(Level.FINER)) {
+                    logger.finer("loaded history by digest " + persistKeyFor(curi)
+                    + " for uri " + curi + " - " + hist);
+                }
+                contentDigestHistory.putAll(hist);
+            }   
+        } catch (Exception e) {
+            logger.log(Level.WARNING, "problem retrieving dedup info from trough segment " + getSegmentId() + " for url " + curi, e);
+        }
+    }
+
+    protected static final String WRITE_SQL_TMPL = 
+            "insert or ignore into dedup (digest_key, url, date, id) values (%s, %s, %s, %s);";
+
+    @Override
+    public void store(CrawlURI curi) {
+        if (!curi.hasContentDigestHistory() || curi.getContentDigestHistory().isEmpty()) {
+            return;
+        }
+        Map<String,Object> hist = curi.getContentDigestHistory();
+
+        try {
+            String digestKey = persistKeyFor(curi);
+            Object url = hist.get(A_ORIGINAL_URL);
+            Date date = ArchiveUtils.parse14DigitISODate((String) hist.get(A_ORIGINAL_DATE), null);
+            Object recordId = hist.get(A_WARC_RECORD_ID);
+            Object[] values = new Object[] { digestKey, url, date, recordId };
+            troughClient().write(getSegmentId(), WRITE_SQL_TMPL, values, SCHEMA_ID);
+        } catch (Exception e) {
+            logger.log(Level.WARNING, "problem writing dedup info to trough segment " + getSegmentId() + " for url " + curi, e);
+
+        }
+    }
+}

contrib/src/main/java/org/archive/trough/TroughClient.java

@@ -0,0 +1,375 @@
+package org.archive.trough;
+
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.net.HttpURLConnection;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.archive.util.ArchiveUtils;
+import org.json.simple.JSONObject;
+import org.json.simple.parser.JSONParser;
+import org.json.simple.parser.ParseException;
+
+import com.rethinkdb.RethinkDB;
+import com.rethinkdb.gen.ast.ReqlExpr;
+import com.rethinkdb.net.Connection;
+
+public class TroughClient {
+    public static String sqlValue(Object o) {
+        if (o == null) {
+            return "null";
+        } else if (o instanceof Date) {
+            return "datetime('" + ArchiveUtils.getLog14Date((Date) o) + "')";
+        } else if (o instanceof Boolean) {
+            if ((Boolean) o) {
+                return "1";
+            } else {
+                return "0";
+            }
+        } else if (o instanceof Number) {
+            return o.toString();
+        } else {
+            // the only character that needs escaped in sqlite string literals
+            // is single-quote, which is escaped as two single-quotes
+            return "'" + o.toString().replaceAll("'", "''") + "'";
+        }
+    }
+
+    private static final Logger logger = Logger.getLogger(TroughClient.class.getName());
+
+    protected static final RethinkDB r = RethinkDB.r;
+
+    protected static final int SIX_HOURS_MS = 6 * 60 * 60 * 1000;
+    protected static final int TEN_MINUTES_MS = 10 * 60 * 1000;
+    protected static final String JSON_MIMETYPE = "application/json";
+    protected static final String SQL_MIMETYPE = "application/sql";
+
+    protected Random rand = new Random();
+    protected Map<String,String> writeUrlCache;
+    protected Map<String,String> readUrlCache;
+    protected Set<String> dirtySegments;
+    protected String[] rethinkServers;
+    protected String rethinkDb;
+    protected Integer promotionInterval;
+
+    public class TroughException extends IOException {
+        private static final long serialVersionUID = 1L;
+        public TroughException(String msg) {
+            super(msg);
+        }
+        public TroughException(Exception e) {
+            super(e);
+        }
+        public TroughException(String msg, Throwable cause) {
+            super(msg, cause);
+        }
+    }
+
+    class Promotrix implements Runnable {
+        @Override
+        public void run() {
+            while (true) {
+                try {
+                    Thread.sleep(promotionInterval * 1000);
+
+                    String[] promoteThese;
+                    synchronized (dirtySegments) {
+                        promoteThese = dirtySegments.toArray(new String[0]);
+                        dirtySegments.clear();
+                    }
+
+                    if (promoteThese.length > 0) {
+                        logger.info("promoting " + promoteThese.length + " trough segments");
+                    }
+                    for (String segmentId: promoteThese) {
+                        try {
+                            promote(segmentId);
+                        } catch (Exception e) {
+                            logger.log(Level.WARNING, "problem promoting segment " + segmentId, e);
+                        }
+                    }
+                } catch (Exception e) {
+                    logger.log(Level.WARNING, "continuing after unexpected exception in promoter thread", e);
+                }
+            }
+        }
+    }
+
+    public TroughClient(String rethinkdbTroughUrl) throws MalformedURLException {
+        this(rethinkdbTroughUrl, null);
+    }
+
+    protected HttpURLConnection httpRequest(String method, String url, String contentType, String payload, int timeout) throws IOException {
+        HttpURLConnection connection = (HttpURLConnection) new URL(url).openConnection();
+
+        connection.setConnectTimeout(timeout);
+        connection.setReadTimeout(timeout);
+
+        connection.setRequestMethod(method);
+
+        if (contentType != null) {
+            connection.setRequestProperty("Content-Type", contentType);
+        }
+        if (payload != null) {
+            connection.setDoOutput(true);
+            byte[] payloadBytes = payload.getBytes("UTF-8");
+            logger.fine(method + " " + url + " " + payloadBytes.length + " bytes " + contentType);
+            connection.setRequestProperty("Content-Length", Integer.toString(payloadBytes.length));
+            connection.getOutputStream().write(payloadBytes);
+            connection.getOutputStream().flush();
+            connection.getOutputStream().close();
+        } else {
+            logger.fine(method + " " + url);
+        }
+
+        return connection;
+    }
+
+    @SuppressWarnings("unchecked")
+    public void promote(String segmentId) throws IOException, TroughException {
+        String url = segmentManagerUrl() + "/promote";
+        JSONObject payload = new JSONObject();
+        payload.put("segment", segmentId);
+
+        HttpURLConnection connection = httpRequest("POST", url, JSON_MIMETYPE, payload.toString(), SIX_HOURS_MS);
+        if (connection.getResponseCode() != 200) {
+            throw new TroughException("received " + connection.getResponseCode() + ": " + connection.getContent()
+            + " in response to POST " + url + " with data " + payload.toString());
+        }
+    }
+
+    /**
+     * Run a rethinkdb query. {@code retries=0} means try once. Default is 9
+     * retries (10 tries).
+     * 
+     * @return result of {@code reql.run()}
+     * @throws TroughException
+     */
+    protected Object rethinkQuery(ReqlExpr reql, Integer retries) throws TroughException {
+        logger.fine("querying rethinkdb: " + reql);
+        if (retries == null || retries < 0) {
+            retries = 9;
+        }
+
+        Exception lastE = null;
+        for (int i = 0; i <= retries; i++) {
+            int whichServer = rand.nextInt(rethinkServers.length);
+            try {
+                String[] hostPort = rethinkServers[whichServer].split(":", 2);
+                String host = hostPort[0];
+                int port = 28015;
+                if (hostPort.length == 2) {
+                    port = Integer.valueOf(hostPort[1]);
+                }
+                Connection conn = r.connection().hostname(host).port(port).db(rethinkDb).connect();
+                Object result = reql.run(conn);
+                return result;
+            } catch (Exception e) {
+                logger.warning("rethinkdb query failed (server=" + rethinkServers[whichServer] + "; " + i + " retries left)");
+                lastE = e;
+            }
+        }
+
+        throw new TroughException(lastE);
+    }
+
+    protected String segmentManagerUrl() throws TroughException {
+        ReqlExpr reql = r.table("services").optArg("read_mode", "majority").getAll("trough-sync-master")
+                .filter(svc -> r.now().sub(svc.g("last_heartbeat")).lt(svc.g("ttl")));
+        @SuppressWarnings("unchecked")
+        Iterable<Map<String, Object>> results = (Iterable<Map<String, Object>>) rethinkQuery(reql, null);
+        for (Map<String,Object> result: results) {
+            return (String) result.get("url");
+        }
+        throw new TroughException("no healthy trough-sync-master in rethinkdb?");
+    }
+
+    /**
+     * 
+     * @param rethinkdbTroughUrl url with schema rethinkdb:// pointing to
+     *          trough configuration database
+     * @param promotionInterval if specified, {@code TroughClient} will spawn a
+     *          thread that "promotes" (pushed to hdfs) "dirty" trough segments
+     *          (segments that have received writes) periodically, sleeping for
+     *          {@code promotionInterval} seconds between cycles 
+     * @throws MalformedURLException 
+     */
+    public TroughClient(String rethinkdbTroughUrl, Integer promotionInterval) throws MalformedURLException {
+        parseRethinkdbUrl(rethinkdbTroughUrl);
+        writeUrlCache = new HashMap<String, String>();
+        readUrlCache = new HashMap<String, String>();
+        dirtySegments = new HashSet<String>();
+
+        if (promotionInterval != null) {
+            this.promotionInterval = promotionInterval;
+            Thread promotrix = new Thread(new Promotrix(), "TroughClient-promotrix");
+            promotrix.setDaemon(true);
+            promotrix.start();
+        }
+    }
+
+    /**
+     * Parses a url like this rethinkdb://server1:port,server2:port/database/table
+     * Sets fields {@code rethinkServers}, {@code rethinkDb}, {@code rethinkTable}
+     * @throws MalformedURLException 
+     */
+    protected void parseRethinkdbUrl(String input) throws MalformedURLException {
+        Matcher m = Pattern.compile("^rethinkdb://([^/]+)/([^/]+)$").matcher(input);
+        if (!m.matches()) {
+            throw new MalformedURLException("failed to parse as rethinkdb url: " + input);
+        }
+
+        rethinkServers = m.group(1).split(",");
+        rethinkDb = m.group(2);
+    }
+
+    // XXX inherits inconsistency in handling exceptions from warcprox trough client
+    @SuppressWarnings("unchecked")
+    public List<Map<String, Object>> read(String segmentId, String sqlTmpl, Object[] values) throws IOException {
+        String readUrl = readUrl(segmentId);
+
+        String[] sqlValues = new String[values.length];
+        for (int i = 0; i < values.length; i++) {
+            sqlValues[i] = sqlValue(values[i]);
+        }
+        String sql = String.format(sqlTmpl, (Object[]) sqlValues);
+
+        HttpURLConnection connection;
+        try {
+            connection = httpRequest("POST", readUrl, SQL_MIMETYPE, sql, TEN_MINUTES_MS);
+            if (connection.getResponseCode() != 200) {
+                readUrlCache.remove(segmentId);
+                throw new TroughException(
+                        "unexpected response" + connection.getResponseCode() + " "
+                        + connection.getResponseMessage() + ": " + connection.getContent()
+                        + " from " + readUrl + " to query: " + sql);
+            }
+
+            Object result = new JSONParser().parse(new InputStreamReader(connection.getInputStream(), "UTF-8"));
+            return (List<Map<String, Object>>) result;
+        } catch (IOException e) {
+            readUrlCache.remove(segmentId);
+            throw e;
+        } catch (ParseException e) {
+            readUrlCache.remove(segmentId);
+            throw new TroughException("problem parsing json response from " + readUrl, e);
+        }
+    }
+
+    protected String readUrl(String segmentId) throws TroughException {
+        if (readUrlCache.get(segmentId) == null) {
+            String url = readUrlNoCache(segmentId);
+            readUrlCache.put(segmentId, url);
+            logger.info("segment " + segmentId + " read url is " + url);
+        }
+        return readUrlCache.get(segmentId);
+    }
+
+    @SuppressWarnings("unchecked")
+    protected String readUrlNoCache(String segmentId) throws TroughException {
+        ReqlExpr reql = r.table("services").getAll(segmentId).optArg("index", "segment")
+                .filter(r.hashMap("role", "trough-read"))
+                .filter(svc -> r.now().sub(svc.g("last_heartbeat")).lt(svc.g("ttl")))
+                .orderBy("load");
+        List<Map<String, Object>> result;
+        result = (List<Map<String,Object>>) rethinkQuery(reql, null);
+        if (result != null && result.size() > 0) {
+            return (String) result.get(0).get("url");
+        } else {
+            throw new TroughException("failed to obtain read url for trough segment " + segmentId);
+        }
+    }
+
+    public void registerSchema(String schemaId, String schemaSql) throws IOException {
+        String url = segmentManagerUrl() + "/schema/" + schemaId + "/sql";
+        HttpURLConnection connection = httpRequest("PUT", url, SQL_MIMETYPE, schemaSql, TEN_MINUTES_MS);
+        if (connection.getResponseCode() != 201 && connection.getResponseCode() != 204) {
+            throw new TroughException("received " + connection.getResponseCode() + ": " + connection.getContent()
+            + " in response to PUT " + url + " with data " + schemaSql);
+        }
+    }
+
+    public void write(String segmentId, String sqlTmpl, Object[] values) throws IOException {
+        write(segmentId, sqlTmpl, values, "default");
+    }
+
+    public void write(String segmentId, String sqlTmpl, Object[] values, String schemaId) throws IOException {
+        String url = writeUrl(segmentId, schemaId);
+
+        String[] sqlValues = new String[values.length];
+        for (int i = 0; i < values.length; i++) {
+            sqlValues[i] = sqlValue(values[i]);
+        }
+        String sql = String.format(sqlTmpl, (Object[]) sqlValues);
+
+        try {
+            HttpURLConnection connection = httpRequest("POST", url, "application/sql", sql, TEN_MINUTES_MS);
+            if (connection.getResponseCode() != 200) {
+                throw new TroughException("unexpected response" + connection.getResponseCode() + " "
+                        + connection.getResponseMessage() + ": " + connection.getContent()
+                        + " from " + url + " to query: " + sql);
+            }            
+            if (!dirtySegments.contains(segmentId)) {
+                synchronized (segmentId) {
+                    dirtySegments.add(segmentId);
+                }
+            }
+        } catch (IOException e) {
+            writeUrlCache.remove("segmentId");
+            throw e;
+        }
+    }
+
+    protected String writeUrl(String segmentId, String schemaId) throws IOException {
+        if (writeUrlCache.get(segmentId) == null) {
+            String url = writeUrlNoCache(segmentId, schemaId);
+            writeUrlCache.put(segmentId, url);
+            logger.info("segment " + segmentId + " write url is " + url);
+        }
+        return writeUrlCache.get(segmentId);
+    }
+
+    @SuppressWarnings("unchecked")
+    protected String writeUrlNoCache(String segmentId, String schemaId) throws IOException {
+        String provisionUrl;
+        String segmentManagerUrl = segmentManagerUrl();
+        if (segmentManagerUrl.endsWith("/")) {
+            provisionUrl = segmentManagerUrl + "provision";
+        } else {
+            provisionUrl = segmentManagerUrl + "/provision";
+        }
+        
+        JSONObject payload = new JSONObject();
+        payload.put("segment", segmentId);
+        payload.put("schema", schemaId);
+        HttpURLConnection connection = httpRequest("POST", provisionUrl, JSON_MIMETYPE, payload.toJSONString(), TEN_MINUTES_MS);
+        if (connection.getResponseCode() != 200) {
+            throw new TroughException("received " + connection.getResponseCode() + ": " + connection.getContent()
+            + " in response to POST " + provisionUrl + " with data " + payload);
+        }
+
+        JSONObject result;
+        try {
+            result = (JSONObject) new JSONParser().parse(new InputStreamReader(connection.getInputStream(), "UTF-8"));
+        } catch (ParseException e) {
+            throw new TroughException("unable to parse response from POST " + provisionUrl + " as json", e);
+        }
+        String writeUrl = (String) result.get("write_url");
+        if (writeUrl == null) {
+            throw new TroughException("write_url missing from response to " + provisionUrl + " - " + result);
+        }
+        return writeUrl;
+    }
+}