diff --git a/htrace-core4/src/main/java/org/apache/htrace/core/Span.java b/htrace-core4/src/main/java/org/apache/htrace/core/Span.java
index e63d414..33908db 100644
--- a/htrace-core4/src/main/java/org/apache/htrace/core/Span.java
+++ b/htrace-core4/src/main/java/org/apache/htrace/core/Span.java
@@ -17,6 +17,7 @@
 package org.apache.htrace.core;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 
@@ -170,8 +171,11 @@
       Map<String, String> traceInfoMap = span.getKVAnnotations();
       if (!traceInfoMap.isEmpty()) {
         jgen.writeObjectFieldStart("n");
-        for (Map.Entry<String, String> e : traceInfoMap.entrySet()) {
-          jgen.writeStringField(e.getKey(), e.getValue());
+        String[] keys = traceInfoMap.keySet().
+            toArray(new String[traceInfoMap.size()]);
+        Arrays.sort(keys);
+        for (String key : keys) {
+          jgen.writeStringField(key, traceInfoMap.get(key));
         }
         jgen.writeEndObject();
       }
diff --git a/htrace-core4/src/test/java/org/apache/htrace/util/TestUtil.java b/htrace-core4/src/test/java/org/apache/htrace/util/TestUtil.java
index 7cb4aed..0869ca0 100644
--- a/htrace-core4/src/test/java/org/apache/htrace/util/TestUtil.java
+++ b/htrace-core4/src/test/java/org/apache/htrace/util/TestUtil.java
@@ -16,8 +16,18 @@
  */
 package org.apache.htrace.util;
 
+import org.apache.htrace.core.MilliSpan;
+import org.apache.htrace.core.Span;
+import org.apache.htrace.core.SpanId;
+import org.apache.htrace.core.TimelineAnnotation;
+
 import java.io.File;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
 import java.util.Map;
+import java.util.Random;
+import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
@@ -88,4 +98,69 @@
       Thread.sleep(periodMs);
     }
   }
+
+  private static long nonZeroRandomLong(Random rand) {
+    long r = 0;
+    do {
+      r = rand.nextLong();
+    } while (r == 0);
+    return r;
+  }
+
+  private static long positiveRandomLong(Random rand) {
+    long r = rand.nextLong();
+    if (r == Long.MIN_VALUE) {
+      // Math.abs can't handle this case
+      return Long.MAX_VALUE;
+    } else if (r > 0) {
+      return r;
+    } else {
+      return -r;
+    }
+  }
+
+  private static String randomString(Random rand) {
+    return new UUID(positiveRandomLong(rand),
+          positiveRandomLong(rand)).toString();
+  }
+
+  public static Span randomSpan(Random rand) {
+    MilliSpan.Builder builder = new MilliSpan.Builder();
+    builder.spanId(
+          new SpanId(nonZeroRandomLong(rand), nonZeroRandomLong(rand)));
+    builder.begin(positiveRandomLong(rand));
+    builder.end(positiveRandomLong(rand));
+    builder.description(randomString(rand));
+    builder.tracerId(randomString(rand));
+    int numParents = rand.nextInt(4);
+    SpanId[] parents = new SpanId[numParents];
+    for (int i = 0; i < numParents; i++) {
+      parents[i] =
+          new SpanId(nonZeroRandomLong(rand), nonZeroRandomLong(rand));
+    }
+    builder.parents(parents);
+    int numTraceInfos = rand.nextInt(4);
+    Map<String, String> traceInfo = new HashMap<String, String>(numTraceInfos);
+    for (int i = 0; i < numTraceInfos; i++) {
+      traceInfo.put(randomString(rand), randomString(rand));
+    }
+    builder.traceInfo(traceInfo);
+    int numTimelineAnnotations = rand.nextInt(4);
+    List<TimelineAnnotation> timeline =
+        new LinkedList<TimelineAnnotation>();
+    for (int i = 0; i < numTimelineAnnotations; i++) {
+      timeline.add(new TimelineAnnotation(positiveRandomLong(rand),
+            randomString(rand)));
+    }
+    builder.timeline(timeline);
+    return builder.build();
+  }
+
+  public static Span[] randomSpans(Random rand, int numSpans) {
+    Span[] spans = new Span[numSpans];
+    for (int i = 0; i < spans.length; i++) {
+      spans[i] = randomSpan(rand);
+    }
+    return spans;
+  }
 }
diff --git a/htrace-htraced/go/src/org/apache/htrace/client/client.go b/htrace-htraced/go/src/org/apache/htrace/client/client.go
index 2ac8a1e..dd3f8a3 100644
--- a/htrace-htraced/go/src/org/apache/htrace/client/client.go
+++ b/htrace-htraced/go/src/org/apache/htrace/client/client.go
@@ -114,32 +114,13 @@
 
 func (hcl *Client) writeSpansHttp(req *common.WriteSpansReq) error {
 	var w bytes.Buffer
-	var err error
-	for i := range req.Spans {
-		var buf []byte
-		buf, err = json.Marshal(req.Spans[i])
-		if err != nil {
-			return errors.New(fmt.Sprintf("Error serializing span: %s",
-				err.Error()))
-		}
-		_, err = w.Write(buf)
-		if err != nil {
-			return errors.New(fmt.Sprintf("Error writing span: %s",
-				err.Error()))
-		}
-		_, err = w.Write([]byte{'\n'})
-		//err = io.WriteString(&w, "\n")
-		if err != nil {
-			return errors.New(fmt.Sprintf("Error writing: %s",
-				err.Error()))
-		}
+	enc := json.NewEncoder(&w)
+	err := enc.Encode(req)
+	if err != nil {
+		return errors.New(fmt.Sprintf("Error serializing span: %s",
+			err.Error()))
 	}
-	customHeaders := make(map[string]string)
-	if req.DefaultTrid != "" {
-		customHeaders["htrace-trid"] = req.DefaultTrid
-	}
-	_, _, err = hcl.makeRestRequest("POST", "writeSpans",
-		&w, customHeaders)
+	_, _, err = hcl.makeRestRequest("POST", "writeSpans", &w)
 	if err != nil {
 		return err
 	}
@@ -182,24 +163,19 @@
 	return spans, nil
 }
 
-var EMPTY = make(map[string]string)
-
 func (hcl *Client) makeGetRequest(reqName string) ([]byte, int, error) {
-	return hcl.makeRestRequest("GET", reqName, nil, EMPTY)
+	return hcl.makeRestRequest("GET", reqName, nil)
 }
 
 // Make a general JSON REST request.
 // Returns the request body, the response code, and the error.
 // Note: if the response code is non-zero, the error will also be non-zero.
-func (hcl *Client) makeRestRequest(reqType string, reqName string, reqBody io.Reader,
-	customHeaders map[string]string) ([]byte, int, error) {
+func (hcl *Client) makeRestRequest(reqType string, reqName string,
+	reqBody io.Reader) ([]byte, int, error) {
 	url := fmt.Sprintf("http://%s/%s",
 		hcl.restAddr, reqName)
 	req, err := http.NewRequest(reqType, url, reqBody)
 	req.Header.Set("Content-Type", "application/json")
-	for k, v := range customHeaders {
-		req.Header.Set(k, v)
-	}
 	client := &http.Client{}
 	resp, err := client.Do(req)
 	if err != nil {
diff --git a/htrace-htraced/go/src/org/apache/htrace/common/rest.go b/htrace-htraced/go/src/org/apache/htrace/common/rest.go
deleted file mode 100644
index b367ed1..0000000
--- a/htrace-htraced/go/src/org/apache/htrace/common/rest.go
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * 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 common
-
-// Info returned by /server/info
-type ServerInfo struct {
-	// The server release version.
-	ReleaseVersion string
-
-	// The git hash that this software was built with.
-	GitVersion string
-}
-
-// Info returned by /server/stats
-type ServerStats struct {
-	Shards []ShardStats
-}
-
-type ShardStats struct {
-	Path string
-
-	// The approximate number of spans present in this shard.  This may be an
-	// underestimate.
-	ApproxNumSpans uint64
-
-	// leveldb.stats information
-	LevelDbStats string
-}
diff --git a/htrace-htraced/go/src/org/apache/htrace/common/rpc.go b/htrace-htraced/go/src/org/apache/htrace/common/rpc.go
index 28521a5..9c7bfad 100644
--- a/htrace-htraced/go/src/org/apache/htrace/common/rpc.go
+++ b/htrace-htraced/go/src/org/apache/htrace/common/rpc.go
@@ -38,14 +38,39 @@
 
 // A request to write spans to htraced.
 type WriteSpansReq struct {
-	DefaultTrid string
+	DefaultTrid string `json:",omitempty"`
 	Spans       []*Span
 }
 
+// Info returned by /server/info
+type ServerInfo struct {
+	// The server release version.
+	ReleaseVersion string
+
+	// The git hash that this software was built with.
+	GitVersion string
+}
+
 // A response to a WriteSpansReq
 type WriteSpansResp struct {
 }
 
+// Info returned by /server/stats
+type ServerStats struct {
+	Shards []ShardStats
+}
+
+type ShardStats struct {
+	Path string
+
+	// The approximate number of spans present in this shard.  This may be an
+	// underestimate.
+	ApproxNumSpans uint64
+
+	// leveldb.stats information
+	LevelDbStats string
+}
+
 // The header which is sent over the wire for HRPC
 type HrpcRequestHeader struct {
 	Magic    uint32
diff --git a/htrace-htraced/go/src/org/apache/htrace/common/span_test.go b/htrace-htraced/go/src/org/apache/htrace/common/span_test.go
index 9de7cee..7fb128d 100644
--- a/htrace-htraced/go/src/org/apache/htrace/common/span_test.go
+++ b/htrace-htraced/go/src/org/apache/htrace/common/span_test.go
@@ -92,7 +92,7 @@
 			End:         5678,
 			Description: "getFileDescriptors",
 			Parents:     []SpanId{},
-			TracerId:   "testTracerId",
+			TracerId:    "testTracerId",
 		}}
 	mh := new(codec.MsgpackHandle)
 	mh.WriteExt = true
diff --git a/htrace-htraced/go/src/org/apache/htrace/conf/config.go b/htrace-htraced/go/src/org/apache/htrace/conf/config.go
index b8f6c84..76af7a6 100644
--- a/htrace-htraced/go/src/org/apache/htrace/conf/config.go
+++ b/htrace-htraced/go/src/org/apache/htrace/conf/config.go
@@ -69,7 +69,7 @@
 }
 
 func getDefaultHTracedConfDir() string {
-	return PATH_SEP + "etc" + PATH_SEP + "htraced" + PATH_SEP + "conf";
+	return PATH_SEP + "etc" + PATH_SEP + "htraced" + PATH_SEP + "conf"
 }
 
 func getHTracedConfDirs(dlog io.Writer) []string {
@@ -78,7 +78,7 @@
 	if len(paths) < 1 {
 		def := getDefaultHTracedConfDir()
 		io.WriteString(dlog, fmt.Sprintf("HTRACED_CONF_DIR defaulting to %s\n", def))
-		return []string{ def }
+		return []string{def}
 	}
 	io.WriteString(dlog, fmt.Sprintf("HTRACED_CONF_DIR=%s\n", confDir))
 	return paths
diff --git a/htrace-htraced/go/src/org/apache/htrace/htrace/cmd.go b/htrace-htraced/go/src/org/apache/htrace/htrace/cmd.go
index 8bc0c64..b5549c5 100644
--- a/htrace-htraced/go/src/org/apache/htrace/htrace/cmd.go
+++ b/htrace-htraced/go/src/org/apache/htrace/htrace/cmd.go
@@ -31,8 +31,8 @@
 	"org/apache/htrace/common"
 	"org/apache/htrace/conf"
 	"os"
-	"time"
 	"strings"
+	"time"
 )
 
 var RELEASE_VERSION string
@@ -126,7 +126,7 @@
 	case serverInfo.FullCommand():
 		os.Exit(printServerInfo(hcl))
 	case serverStats.FullCommand():
-		if (*serverStatsJson) {
+		if *serverStatsJson {
 			os.Exit(printServerStatsJson(hcl))
 		} else {
 			os.Exit(printServerStats(hcl))
@@ -195,7 +195,7 @@
 	}
 	fmt.Printf("HTraced server stats:\n")
 	fmt.Printf("%d leveldb shards.\n", len(stats.Shards))
-	for i := range(stats.Shards) {
+	for i := range stats.Shards {
 		shard := stats.Shards[i]
 		fmt.Printf("==== %s ===\n", shard.Path)
 		fmt.Printf("Approximate number of spans: %d\n", shard.ApproxNumSpans)
diff --git a/htrace-htraced/go/src/org/apache/htrace/htraced/datastore.go b/htrace-htraced/go/src/org/apache/htrace/htraced/datastore.go
index 0595d36..9fb9920 100644
--- a/htrace-htraced/go/src/org/apache/htrace/htraced/datastore.go
+++ b/htrace-htraced/go/src/org/apache/htrace/htraced/datastore.go
@@ -953,19 +953,19 @@
 }
 
 func (store *dataStore) ServerStats() *common.ServerStats {
-	serverStats := common.ServerStats {
-		Shards : make([]common.ShardStats, len(store.shards)),
+	serverStats := common.ServerStats{
+		Shards: make([]common.ShardStats, len(store.shards)),
 	}
-	for shardIdx := range(store.shards) {
+	for shardIdx := range store.shards {
 		shard := store.shards[shardIdx]
 		serverStats.Shards[shardIdx].Path = shard.path
-		r := levigo.Range {
-			Start : append([]byte{SPAN_ID_INDEX_PREFIX},
+		r := levigo.Range{
+			Start: append([]byte{SPAN_ID_INDEX_PREFIX},
 				common.INVALID_SPAN_ID.Val()...),
-			Limit : append([]byte{SPAN_ID_INDEX_PREFIX + 1},
+			Limit: append([]byte{SPAN_ID_INDEX_PREFIX + 1},
 				common.INVALID_SPAN_ID.Val()...),
 		}
-		vals := shard.ldb.GetApproximateSizes([]levigo.Range { r })
+		vals := shard.ldb.GetApproximateSizes([]levigo.Range{r})
 		serverStats.Shards[shardIdx].ApproxNumSpans = vals[0]
 		serverStats.Shards[shardIdx].LevelDbStats =
 			shard.ldb.PropertyValue("leveldb.stats")
diff --git a/htrace-htraced/go/src/org/apache/htrace/htraced/rest.go b/htrace-htraced/go/src/org/apache/htrace/htraced/rest.go
index 97b2bca..cbfc508 100644
--- a/htrace-htraced/go/src/org/apache/htrace/htraced/rest.go
+++ b/htrace-htraced/go/src/org/apache/htrace/htraced/rest.go
@@ -193,37 +193,24 @@
 	} else {
 		dec = json.NewDecoder(req.Body)
 	}
-	spans := make([]*common.Span, 0, 32)
-	defaultTrid := req.Header.Get("htrace-trid")
-	for {
-		var span common.Span
-		err := dec.Decode(&span)
-		if err != nil {
-			if err != io.EOF {
-				writeError(hand.lg, w, http.StatusBadRequest,
-					fmt.Sprintf("Error parsing spans: %s", err.Error()))
-				return
-			}
-			break
-		}
-		spanIdProblem := span.Id.FindProblem()
-		if spanIdProblem != "" {
-			writeError(hand.lg, w, http.StatusBadRequest,
-				fmt.Sprintf("Invalid span ID: %s", spanIdProblem))
-			return
-		}
-		if span.TracerId == "" {
-			span.TracerId = defaultTrid
-		}
-		spans = append(spans, &span)
+	var msg common.WriteSpansReq
+	err := dec.Decode(&msg)
+	if (err != nil) && (err != io.EOF) {
+		writeError(hand.lg, w, http.StatusBadRequest,
+			fmt.Sprintf("Error parsing WriteSpansReq: %s", err.Error()))
+		return
 	}
 	hand.lg.Debugf("writeSpansHandler: received %d span(s).  defaultTrid = %s\n",
-		len(spans), defaultTrid)
-	for spanIdx := range spans {
+		len(msg.Spans), msg.DefaultTrid)
+	for spanIdx := range msg.Spans {
 		if hand.lg.DebugEnabled() {
-			hand.lg.Debugf("writing span %s\n", spans[spanIdx].ToJson())
+			hand.lg.Debugf("writing span %s\n", msg.Spans[spanIdx].ToJson())
 		}
-		hand.store.WriteSpan(spans[spanIdx])
+		span := msg.Spans[spanIdx]
+		if span.TracerId == "" {
+			span.TracerId = msg.DefaultTrid
+		}
+		hand.store.WriteSpan(span)
 	}
 }
 
diff --git a/htrace-htraced/pom.xml b/htrace-htraced/pom.xml
index 88fd2fc..4478dc5 100644
--- a/htrace-htraced/pom.xml
+++ b/htrace-htraced/pom.xml
@@ -86,6 +86,10 @@
                   <pattern>org.eclipse.jetty</pattern>
                   <shadedPattern>org.apache.htrace.shaded.jetty</shadedPattern>
                 </relocation>
+                <dependency>
+                  <pattern>org.msgpack</pattern>
+                  <shadedPattern>org.apache.htrace.msgpack</shadedPattern>
+                </dependency>
               </relocations>
             </configuration>
             <goals>
@@ -200,12 +204,11 @@
       <groupId>com.fasterxml.jackson.core</groupId>
       <artifactId>jackson-databind</artifactId>
     </dependency>
-    <!-- htraced rest client deps. -->
-    <!--Is this too much? Pulls down jetty-http, jetty-server, jetty-io
-     This is new-style jetty client, jetty9 and jdk7 required.
-     It can do async but we will use it synchronously at first.
-     Has nice tutorial: http://www.eclipse.org/jetty/documentation/9.1.5.v20140505/http-client-api.html
-     -->
+    <dependency>
+      <groupId>org.msgpack</groupId>
+      <artifactId>msgpack-core</artifactId>
+      <version>0.7.0-p9</version>
+    </dependency>
     <dependency>
       <groupId>org.eclipse.jetty</groupId>
       <artifactId>jetty-client</artifactId>
diff --git a/htrace-htraced/src/main/java/org/apache/htrace/impl/BufferManager.java b/htrace-htraced/src/main/java/org/apache/htrace/impl/BufferManager.java
new file mode 100644
index 0000000..29816eb
--- /dev/null
+++ b/htrace-htraced/src/main/java/org/apache/htrace/impl/BufferManager.java
@@ -0,0 +1,74 @@
+/**
+ * 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.htrace.impl;
+
+import java.io.IOException;
+
+import org.apache.htrace.core.Span;
+
+/**
+ * A buffer which contains span data and is able to send it over the network.
+ *
+ * BufferManager functions are not thread-safe.  You must rely on external
+ * synchronization to protect buffers from concurrent operations.
+ */
+interface BufferManager {
+  /**
+   * Write a span to this buffer.
+   *
+   * @param span            The span to write.
+   *
+   * @throws IOException    If the buffer doesn't have enough space to hold the
+   *                          new span.  We will not write a partial span to the
+   *                          buffer in this case.
+   */
+  void writeSpan(Span span) throws IOException;
+
+  /**
+   * Get the amount of content currently in the buffer.
+   */
+  int contentLength();
+
+  /**
+   * Get the number of spans currently buffered.
+   */
+  int getNumberOfSpans();
+
+  /**
+   * Prepare the buffers to be flushed.
+   */
+  void prepare() throws IOException;
+
+  /**
+   * Flush this buffer to htraced.
+   *
+   * This is a blocking operation which will not return until the buffer is
+   * completely flushed.
+   */
+  void flush() throws IOException;
+
+  /**
+   * Clear the data in this buffer.
+   */
+  void clear();
+
+  /**
+   * Closes the buffer manager and frees all resources.
+   */
+  void close();
+}
diff --git a/htrace-htraced/src/main/java/org/apache/htrace/impl/Conf.java b/htrace-htraced/src/main/java/org/apache/htrace/impl/Conf.java
new file mode 100644
index 0000000..cdd176f
--- /dev/null
+++ b/htrace-htraced/src/main/java/org/apache/htrace/impl/Conf.java
@@ -0,0 +1,356 @@
+/**
+ * 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.htrace.impl;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import org.apache.htrace.core.HTraceConfiguration;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+/**
+ * The configuration of the HTracedSpanReceiver.
+ *
+ * This class extracts all the relevant configuration information for
+ * HTracedSpanReceiver from the HTraceConfiguration.  It performs parsing and
+ * bounds-checking for the configuration keys.
+ *
+ * It is more efficient to store the configuration as final values in this
+ * structure than to access the HTraceConfiguration object directly.  This is
+ * especially true when the HTraceConfiguration object is a thin shim around a
+ * Hadoop Configuration object, which requires synchronization to access.
+ */
+class Conf {
+  private static final Log LOG = LogFactory.getLog(Conf.class);
+
+  /**
+   * Address of the htraced server.
+   */
+  final static String ADDRESS_KEY =
+      "htraced.receiver.address";
+
+  /**
+   * The minimum number of milliseconds to wait for a read or write
+   * operation on the network.
+   */
+  final static String IO_TIMEOUT_MS_KEY =
+      "htraced.receiver.io.timeout.ms";
+  final static int IO_TIMEOUT_MS_DEFAULT = 60000;
+  final static int IO_TIMEOUT_MS_MIN = 50;
+
+  /**
+   * The minimum number of milliseconds to wait for a network
+   * connection attempt.
+   */
+  final static String CONNECT_TIMEOUT_MS_KEY =
+      "htraced.receiver.connect.timeout.ms";
+  final static int CONNECT_TIMEOUT_MS_DEFAULT = 60000;
+  final static int CONNECT_TIMEOUT_MS_MIN = 50;
+
+  /**
+   * The minimum number of milliseconds to keep alive a connection when it's
+   * not in use.
+   */
+  final static String IDLE_TIMEOUT_MS_KEY =
+      "htraced.receiver.idle.timeout.ms";
+  final static int IDLE_TIMEOUT_MS_DEFAULT = 60000;
+  final static int IDLE_TIMEOUT_MS_MIN = 0;
+
+  /**
+   * Configure the retry times to use when an attempt to flush spans to
+   * htraced fails.  This is configured as a comma-separated list of delay
+   * times in milliseconds.  If the configured value is empty, no retries
+   * will be made.
+   */
+  final static String FLUSH_RETRY_DELAYS_KEY =
+      "htraced.flush.retry.delays.key";
+  final static String FLUSH_RETRY_DELAYS_DEFAULT =
+      "1000,30000";
+
+  /**
+   * The maximum length of time to go in between flush attempts.
+   * Once this time elapses, a flush will be triggered even if we don't
+   * have that many spans buffered.
+   */
+  final static String MAX_FLUSH_INTERVAL_MS_KEY =
+      "htraced.receiver.max.flush.interval.ms";
+  final static int MAX_FLUSH_INTERVAL_MS_DEFAULT = 60000;
+  final static int MAX_FLUSH_INTERVAL_MS_MIN = 10;
+
+  /**
+   * Whether or not to use msgpack for span serialization.
+   * If this key is false, JSON over REST will be used.
+   * If this key is true, msgpack over custom RPC will be used.
+   */
+  final static String PACKED_KEY =
+      "htraced.receiver.packed";
+  final static boolean PACKED_DEFAULT = true;
+
+  /**
+   * The size of the span buffers.
+   */
+  final static String BUFFER_SIZE_KEY =
+      "htraced.receiver.buffer.size";
+  final static int BUFFER_SIZE_DEFAULT = 48 * 1024 * 1024;
+  static int BUFFER_SIZE_MIN = 4 * 1024 * 1024;
+  // The maximum buffer size should not be longer than
+  // PackedBuffer.MAX_HRPC_BODY_LENGTH.
+  final static int BUFFER_SIZE_MAX = 63 * 1024 * 1024;
+
+  /**
+   * Set the fraction of the span buffer which needs to fill up before we
+   * will automatically trigger a flush.  This is a fraction, not a percentage.
+   * It is between 0 and 1.
+   */
+  final static String BUFFER_SEND_TRIGGER_FRACTION_KEY =
+      "htraced.receiver.buffer.send.trigger.fraction";
+  final static double BUFFER_SEND_TRIGGER_FRACTION_DEFAULT = 0.5;
+  final static double BUFFER_SEND_TRIGGER_FRACTION_MIN = 0.1;
+
+  /**
+   * The length of time which receiveSpan should wait for a free spot in a
+   * span buffer before giving up and dropping the span
+   */
+  final static String SPAN_DROP_TIMEOUT_MS_KEY =
+      "htraced.max.buffer.full.retry.ms.key";
+  final static int SPAN_DROP_TIMEOUT_MS_DEFAULT = 5000;
+
+  /**
+   * The length of time we should wait between displaying log messages on the
+   * rate-limited loggers.
+   */
+  final static String ERROR_LOG_PERIOD_MS_KEY =
+      "htraced.error.log.period.ms";
+  final static long ERROR_LOG_PERIOD_MS_DEFAULT = 30000L;
+
+  @JsonProperty("ioTimeoutMs")
+  final int ioTimeoutMs;
+
+  @JsonProperty("connectTimeoutMs")
+  final int connectTimeoutMs;
+
+  @JsonProperty("idleTimeoutMs")
+  final int idleTimeoutMs;
+
+  @JsonProperty("flushRetryDelays")
+  final int[] flushRetryDelays;
+
+  @JsonProperty("maxFlushIntervalMs")
+  final int maxFlushIntervalMs;
+
+  @JsonProperty("packed")
+  final boolean packed;
+
+  @JsonProperty("bufferSize")
+  final int bufferSize;
+
+  @JsonProperty("spanDropTimeoutMs")
+  final int spanDropTimeoutMs;
+
+  @JsonProperty("errorLogPeriodMs")
+  final long errorLogPeriodMs;
+
+  @JsonProperty("triggerSize")
+  final int triggerSize;
+
+  @JsonProperty("endpointStr")
+  final String endpointStr;
+
+  @JsonProperty("endpoint")
+  final InetSocketAddress endpoint;
+
+  private static int getBoundedInt(final HTraceConfiguration conf,
+        String key, int defaultValue, int minValue, int maxValue) {
+    int val = conf.getInt(key, defaultValue);
+    if (val < minValue) {
+      LOG.warn("Can't set " + key + " to " + val + ".  Using minimum value " +
+          "of " + minValue + " instead.");
+      return minValue;
+    } else if (val > maxValue) {
+      LOG.warn("Can't set " + key + " to " + val + ".  Using maximum value " +
+          "of " + maxValue + " instead.");
+      return maxValue;
+    }
+    return val;
+  }
+
+  private static long getBoundedLong(final HTraceConfiguration conf,
+        String key, long defaultValue, long minValue, long maxValue) {
+    String strVal = conf.get(key, Long.toString(defaultValue));
+    long val = 0;
+    try {
+      val = Long.parseLong(strVal);
+    } catch (NumberFormatException nfe) {
+      throw new IllegalArgumentException("Bad value for '" + key +
+        "': should be long");
+    }
+    if (val < minValue) {
+      LOG.warn("Can't set " + key + " to " + val + ".  Using minimum value " +
+          "of " + minValue + " instead.");
+      return minValue;
+    } else if (val > maxValue) {
+      LOG.warn("Can't set " + key + " to " + val + ".  Using maximum value " +
+          "of " + maxValue + " instead.");
+      return maxValue;
+    }
+    return val;
+  }
+
+  private static double getBoundedDouble(final HTraceConfiguration conf,
+        String key, double defaultValue, double minValue, double maxValue) {
+    String strVal = conf.get(key, Double.toString(defaultValue));
+    double val = 0;
+    try {
+      val = Double.parseDouble(strVal);
+    } catch (NumberFormatException nfe) {
+      throw new IllegalArgumentException("Bad value for '" + key +
+        "': should be double");
+    }
+    if (val < minValue) {
+      LOG.warn("Can't set " + key + " to " + val + ".  Using minimum value " +
+          "of " + minValue + " instead.");
+      return minValue;
+    }
+    if (val > maxValue) {
+      LOG.warn("Can't set " + key + " to " + val + ".  Using maximum value " +
+          "of " + maxValue + " instead.");
+      return maxValue;
+    }
+    return val;
+  }
+
+  private static int parseColonPort(String portStr) throws IOException {
+    int colonPosition = portStr.indexOf(':');
+    if (colonPosition != 0) {
+      throw new IOException("Invalid port string " + portStr);
+    }
+    int port = Integer.parseInt(portStr.substring(1), 10);
+    if ((port < 0) || (port > 65535)) {
+      throw new IOException("Invalid port number " + port);
+    }
+    return port;
+  }
+
+  /**
+   * Parse a hostname:port or ip:port pair.
+   *
+   * @param str       The string to parse.
+   * @return          The socket address.
+   */
+  InetSocketAddress parseHostPortPair(String str) throws IOException {
+    str = str.trim();
+    if (str.isEmpty()) {
+      throw new IOException("No hostname:port pair given.");
+    }
+    int bracketBegin = str.indexOf('[');
+    if (bracketBegin == 0) {
+      // Parse an ipv6-style address enclosed in square brackets.
+      int bracketEnd = str.indexOf(']');
+      if (bracketEnd < 0) {
+        throw new IOException("Found left bracket, but no corresponding " +
+            "right bracket, in " + str);
+      }
+      String host = str.substring(bracketBegin + 1, bracketEnd);
+      int port = parseColonPort(str.substring(bracketEnd + 1));
+      return InetSocketAddress.createUnresolved(host, port);
+    } else if (bracketBegin > 0) {
+        throw new IOException("Found a left bracket that wasn't at the " +
+          "start of the host:port pair in " + str);
+    } else {
+      int colon = str.indexOf(':');
+      if (colon <= 0) {
+        throw new IOException("No port component found in " + str);
+      }
+      String host = str.substring(0, colon);
+      int port = parseColonPort(str.substring(colon));
+      return InetSocketAddress.createUnresolved(host, port);
+    }
+  }
+
+  static int[] getIntArray(String arrayStr) {
+    String[] array = arrayStr.split(",");
+    int nonEmptyEntries = 0;
+    for (String str : array) {
+      if (!str.trim().isEmpty()) {
+        nonEmptyEntries++;
+      }
+    }
+    int[] ret = new int[nonEmptyEntries];
+    int i = 0;
+    for (String str : array) {
+      if (!str.trim().isEmpty()) {
+        ret[i++] = Integer.parseInt(str);
+      }
+    }
+    return ret;
+  }
+
+  Conf(HTraceConfiguration conf) throws IOException {
+    this.ioTimeoutMs = getBoundedInt(conf, IO_TIMEOUT_MS_KEY,
+              IO_TIMEOUT_MS_DEFAULT,
+              IO_TIMEOUT_MS_MIN, Integer.MAX_VALUE);
+    this.connectTimeoutMs = getBoundedInt(conf, CONNECT_TIMEOUT_MS_KEY,
+              CONNECT_TIMEOUT_MS_DEFAULT,
+              CONNECT_TIMEOUT_MS_MIN, Integer.MAX_VALUE);
+    this.idleTimeoutMs = getBoundedInt(conf, IDLE_TIMEOUT_MS_KEY,
+              IDLE_TIMEOUT_MS_DEFAULT,
+              IDLE_TIMEOUT_MS_MIN, Integer.MAX_VALUE);
+    this.flushRetryDelays = getIntArray(conf.get(FLUSH_RETRY_DELAYS_KEY,
+              FLUSH_RETRY_DELAYS_DEFAULT));
+    this.maxFlushIntervalMs = getBoundedInt(conf, MAX_FLUSH_INTERVAL_MS_KEY,
+              MAX_FLUSH_INTERVAL_MS_DEFAULT,
+              MAX_FLUSH_INTERVAL_MS_MIN, Integer.MAX_VALUE);
+    this.packed = conf.getBoolean(PACKED_KEY, PACKED_DEFAULT);
+    this.bufferSize = getBoundedInt(conf, BUFFER_SIZE_KEY,
+              BUFFER_SIZE_DEFAULT,
+              BUFFER_SIZE_MIN, BUFFER_SIZE_MAX);
+    double triggerFraction = getBoundedDouble(conf,
+              BUFFER_SEND_TRIGGER_FRACTION_KEY,
+              BUFFER_SEND_TRIGGER_FRACTION_DEFAULT,
+              BUFFER_SEND_TRIGGER_FRACTION_MIN, 1.0);
+    this.spanDropTimeoutMs = conf.getInt(SPAN_DROP_TIMEOUT_MS_KEY,
+        SPAN_DROP_TIMEOUT_MS_DEFAULT);
+    this.errorLogPeriodMs = getBoundedLong(conf, ERROR_LOG_PERIOD_MS_KEY,
+        ERROR_LOG_PERIOD_MS_DEFAULT, 0, Long.MAX_VALUE);
+    this.triggerSize = (int)(this.bufferSize * triggerFraction);
+    try {
+      this.endpointStr = conf.get(ADDRESS_KEY, "");
+      this.endpoint = parseHostPortPair(endpointStr);
+    } catch (IOException e) {
+      throw new IOException("Error reading " + ADDRESS_KEY + ": " +
+          e.getMessage());
+    }
+  }
+
+  @Override
+  public String toString() {
+    ObjectMapper mapper = new ObjectMapper();
+    mapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
+    try {
+      return mapper.writeValueAsString(this);
+    } catch (JsonProcessingException e) {
+      throw new RuntimeException(e);
+    }
+  }
+}
diff --git a/htrace-htraced/src/main/java/org/apache/htrace/impl/HTracedRESTReceiver.java b/htrace-htraced/src/main/java/org/apache/htrace/impl/HTracedRESTReceiver.java
deleted file mode 100644
index 643bbd5..0000000
--- a/htrace-htraced/src/main/java/org/apache/htrace/impl/HTracedRESTReceiver.java
+++ /dev/null
@@ -1,431 +0,0 @@
-/**
- * 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.htrace.impl;
-
-import java.io.IOException;
-import java.net.URL;
-import java.util.ArrayDeque;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.ReentrantLock;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.htrace.core.HTraceConfiguration;
-import org.apache.htrace.core.Span;
-import org.apache.htrace.core.SpanReceiver;
-import org.eclipse.jetty.client.HttpClient;
-import org.eclipse.jetty.client.api.ContentResponse;
-import org.eclipse.jetty.client.api.Request;
-import org.eclipse.jetty.client.util.StringContentProvider;
-import org.eclipse.jetty.http.HttpField;
-import org.eclipse.jetty.http.HttpHeader;
-import org.eclipse.jetty.http.HttpMethod;
-import org.eclipse.jetty.http.HttpStatus;
-
-/**
- * A {@link SpanReceiver} that passes Spans to htraced via REST. Implementation minimizes
- * dependencies and aims for small footprint since this client will be the guest of another,
- * the process traced.
- *
- * <p>Logs via commons-logging. Uses jetty client. Jetty has its own logging. To connect, see
- * jetty logging to commons-logging and see https://issues.apache.org/jira/browse/HADOOP-6807
- * and http://blogs.bytecode.com.au/glen/2005/06/21/getting-your-logging-working-in-jetty.html.
- *
- * <p>This client depends on the REST defined in <code>rest.go</code> in the htraced REST server.
- *
- * <p>Create an instance by doing:
- * <code>SpanReceiver receiver = new HTracedRESTReceiver(conf);</code> where conf is an instance
- * of {@link HTraceConfiguration}. See the public keys defined below for what we will look for in
- * the configuration.  For example, set {@link #HTRACED_REST_URL_KEY} if
- * <code>htraced</code> is in a non-standard location. Then call
- * <code>receiver.receiveSpan(Span);</code> to send spans to an htraced
- * instance. This method returns immediately. It sends the spans in background.
- *
- * <p>TODO: Shading works?
- * TODO: Add lazy start; don't start background thread till a span gets queued.
- * TODO: Add some metrics; how many times we've run, how many spans and what size we've sent.
- */
-public class HTracedRESTReceiver extends SpanReceiver {
-  private static final Log LOG = LogFactory.getLog(HTracedRESTReceiver.class);
-
-  /**
-   * The HttpClient to use for this receiver.
-   */
-  private final HttpClient httpClient;
-
-  /**
-   * The maximum number of spans to buffer.
-   */
-  private final int capacity;
-
-  /**
-   * REST URL to use writing Spans.
-   */
-  private final String url;
-
-  /**
-   * The maximum number of spans to send in a single PUT.
-   */
-  private final int maxToSendAtATime;
-
-  /**
-   * Runs background task to do the REST PUT.
-   */
-  private final PostSpans postSpans;
-
-  /**
-   * Thread for postSpans
-   */
-  private final Thread postSpansThread;
-
-  /**
-   * The connection timeout in milliseconds.
-   */
-  public static final String CLIENT_CONNECT_TIMEOUT_MS_KEY = "client.connect.timeout.ms";
-  private static final int CLIENT_CONNECT_TIMEOUT_MS_DEFAULT = 30000;
-
-  /**
-   * The idle timeout in milliseconds.
-   */
-  public static final String CLIENT_IDLE_TIMEOUT_MS_KEY = "client.idle.timeout.ms";
-  private static final int CLIENT_IDLE_TIMEOUT_MS_DEFAULT = 120000;
-
-  /**
-   * URL of the htraced REST server we are to talk to.
-   */
-  public static final String HTRACED_REST_URL_KEY = "htraced.rest.url";
-  private static final String HTRACED_REST_URL_DEFAULT = "http://localhost:9095/";
-
-  /**
-   * Maximum size of the queue to accumulate spans in.
-   * Cleared by the background thread that does the REST POST to htraced.
-   */
-  public static final String CLIENT_REST_QUEUE_CAPACITY_KEY = "client.rest.queue.capacity";
-  private static final int CLIENT_REST_QUEUE_CAPACITY_DEFAULT = 1000000;
-
-  /**
-   * Period at which the background thread that does the REST POST to htraced in ms.
-   */
-  public static final String CLIENT_REST_PERIOD_MS_KEY = "client.rest.period.ms";
-  private static final int CLIENT_REST_PERIOD_MS_DEFAULT = 30000;
-
-  /**
-   * Maximum spans to post to htraced at a time.
-   */
-  public static final String CLIENT_REST_MAX_SPANS_AT_A_TIME_KEY =
-    "client.rest.batch.size";
-  private static final int CLIENT_REST_MAX_SPANS_AT_A_TIME_DEFAULT = 100;
-
-  /**
-   * Lock protecting the PostSpans data.
-   */
-  private ReentrantLock lock = new ReentrantLock();
-
-  /**
-   * Condition variable used to wake up the PostSpans thread.
-   */
-  private Condition cond = lock.newCondition();
-
-  /**
-   * True if we should shut down.
-   * Protected by the lock.
-   */
-  private boolean shutdown = false;
-
-  /**
-   * Simple bounded queue to hold spans between periodic runs of the httpclient.
-   * Protected by the lock.
-   */
-  private final ArrayDeque<Span> spans;
-
-  /**
-   * Keep last time we logged we were at capacity; used to prevent flooding of logs with
-   * "at capacity" messages.
-   */
-  private AtomicLong lastAtCapacityWarningLog = new AtomicLong(0L);
-
-  /**
-   * True if we should flush as soon as possible.  Protected by the lock.
-   */
-  private boolean mustStartFlush;
-
-  /**
-   * Create an HttpClient instance.
-   *
-   * @param connTimeout         The timeout to use for connecting.
-   * @param idleTimeout         The idle timeout to use.
-   */
-  HttpClient createHttpClient(long connTimeout, long idleTimeout) {
-    HttpClient httpClient = new HttpClient();
-    httpClient.setUserAgentField(new HttpField(HttpHeader.USER_AGENT,
-      this.getClass().getSimpleName()));
-    httpClient.setConnectTimeout(connTimeout);
-    httpClient.setIdleTimeout(idleTimeout);
-    return httpClient;
-  }
-
-  /**
-   * Constructor.
-   * You must call {@link #close()} post construction when done.
-   * @param conf
-   * @throws Exception
-   */
-  public HTracedRESTReceiver(final HTraceConfiguration conf) throws Exception {
-    int connTimeout = conf.getInt(CLIENT_CONNECT_TIMEOUT_MS_KEY,
-                                  CLIENT_CONNECT_TIMEOUT_MS_DEFAULT);
-    int idleTimeout = conf.getInt(CLIENT_IDLE_TIMEOUT_MS_KEY,
-                                  CLIENT_IDLE_TIMEOUT_MS_DEFAULT);
-    this.httpClient = createHttpClient(connTimeout, idleTimeout);
-    this.capacity = conf.getInt(CLIENT_REST_QUEUE_CAPACITY_KEY, CLIENT_REST_QUEUE_CAPACITY_DEFAULT);
-    this.spans = new ArrayDeque<Span>(capacity);
-    // Build up the writeSpans URL.
-    URL restServer = new URL(conf.get(HTRACED_REST_URL_KEY, HTRACED_REST_URL_DEFAULT));
-    URL url = new URL(restServer.getProtocol(), restServer.getHost(), restServer.getPort(), "/writeSpans");
-    this.url = url.toString();
-    // Period at which we run the background thread that does the REST POST to htraced.
-    int periodInMs = conf.getInt(CLIENT_REST_PERIOD_MS_KEY, CLIENT_REST_PERIOD_MS_DEFAULT);
-    // Maximum spans to send in one go
-    this.maxToSendAtATime =
-      conf.getInt(CLIENT_REST_MAX_SPANS_AT_A_TIME_KEY, CLIENT_REST_MAX_SPANS_AT_A_TIME_DEFAULT);
-    // Start up the httpclient.
-    this.httpClient.start();
-    // Start the background thread.
-    this.postSpans = new PostSpans(periodInMs);
-    this.postSpansThread = new Thread(postSpans);
-    this.postSpansThread.setDaemon(true);
-    this.postSpansThread.setName("PostSpans");
-    this.postSpansThread.start();
-    if (LOG.isDebugEnabled()) {
-      LOG.debug("Created new HTracedRESTReceiver with connTimeout=" +
-            connTimeout + ", idleTimeout = " + idleTimeout + ", capacity=" +
-            capacity + ", url=" + url +  ", periodInMs=" + periodInMs +
-            ", maxToSendAtATime=" + maxToSendAtATime);
-    }
-  }
-
-  /**
-   * POST spans runnable.
-   * Run on a period. Services the passed in queue taking spans and sending them to traced via http.
-   */
-  private class PostSpans implements Runnable {
-    private final long periodInNs;
-    private final ArrayDeque<Span> spanBuf;
-
-    private PostSpans(long periodInMs) {
-      this.periodInNs = TimeUnit.NANOSECONDS.
-          convert(periodInMs, TimeUnit.MILLISECONDS);
-      this.spanBuf = new ArrayDeque<Span>(maxToSendAtATime);
-    }
-
-    /**
-     * The span sending thread.
-     *
-     * We send a batch of spans for one of two reasons: there are already
-     * maxToSendAtATime spans in the buffer, or the client.rest.period.ms
-     * has elapsed.  The idea is that we want to strike a balance between
-     * sending a lot of spans at a time, for efficiency purposes, and
-     * making sure that we don't buffer spans locally for too long.
-     *
-     * The longer we buffer spans locally, the longer we will have to wait
-     * to see the results of our client operations in the GUI, and the higher
-     * the risk of losing them if the client crashes.
-     */
-    @Override
-    public void run() {
-      long waitNs;
-      try {
-        waitNs = periodInNs;
-        while (true) {
-          lock.lock();
-          try {
-            if (shutdown) {
-              if (spans.isEmpty()) {
-                LOG.debug("Shutting down PostSpans thread...");
-                break;
-              }
-            } else {
-              try {
-                waitNs = cond.awaitNanos(waitNs);
-                if (mustStartFlush) {
-                  waitNs = 0;
-                  mustStartFlush = false;
-                }
-              } catch (InterruptedException e) {
-                LOG.info("Got InterruptedException");
-                waitNs = 0;
-              }
-            }
-            if ((spans.size() > maxToSendAtATime) || (waitNs <= 0) ||
-                    shutdown) {
-              loadSpanBuf();
-              waitNs = periodInNs;
-            }
-          } finally {
-            lock.unlock();
-          }
-          // Once the lock has been safely released, we can do some network
-          // I/O without blocking the client process.
-          if (!spanBuf.isEmpty()) {
-            sendSpans();
-            spanBuf.clear();
-          }
-        }
-      } finally {
-        if (httpClient != null) {
-          try {
-            httpClient.stop();
-          } catch (Exception e) {
-            LOG.error("Error shutting down httpClient", e);
-          }
-        }
-        spans.clear();
-      }
-    }
-
-    private void loadSpanBuf() {
-      for (int loaded = 0; loaded < maxToSendAtATime; loaded++) {
-        Span span = spans.pollFirst();
-        if (span == null) {
-          return;
-        }
-        spanBuf.add(span);
-      }
-    }
-
-    private void sendSpans() {
-      try {
-        Request request = httpClient.newRequest(url).method(HttpMethod.POST);
-        request.header(HttpHeader.CONTENT_TYPE, "application/json");
-        StringBuilder bld = new StringBuilder();
-        for (Span span : spanBuf) {
-          bld.append(span.toJson());
-        }
-        request.content(new StringContentProvider(bld.toString()));
-        ContentResponse response = request.send();
-        if (response.getStatus() == HttpStatus.OK_200) {
-          if (LOG.isDebugEnabled()) {
-            LOG.debug("POSTED " + spanBuf.size() + " spans");
-          }
-        } else {
-          LOG.error("Status: " + response.getStatus());
-          LOG.error(response.getHeaders());
-          LOG.error(response.getContentAsString());
-        }
-      } catch (InterruptedException e) {
-        LOG.error(e);
-      } catch (TimeoutException e) {
-        LOG.error(e);
-      } catch (ExecutionException e) {
-        LOG.error(e);
-      }
-    }
-  }
-
-  @Override
-  public void close() throws IOException {
-    LOG.debug("Closing HTracedRESTReceiver(" + url + ").");
-    lock.lock();
-    try {
-      this.shutdown = true;
-      cond.signal();
-    } finally {
-      lock.unlock();
-    }
-    try {
-      postSpansThread.join(120000);
-      if (postSpansThread.isAlive()) {
-        LOG.error("Timed out without closing HTracedRESTReceiver(" +
-                  url + ").");
-      } else {
-        LOG.debug("Closed HTracedRESTReceiver(" + url + ").");
-      }
-    } catch (InterruptedException e) {
-      LOG.error("Interrupted while joining postSpans", e);
-    }
-  }
-
-  /**
-   * Start flushing the buffered spans.
-   *
-   * Note that even after calling this function, you will still have to wait
-   * for the flush to finish happening.  This function just starts the flush;
-   * it does not block until it has completed.  You also do not get
-   * "read-after-write consistency" with htraced... the spans that are
-   * written may be buffered for a short period of time prior to being
-   * readable.  This is not a problem for production use (since htraced is not
-   * a database), but it means that most unit tests will need a loop in their
-   * "can I read what I wrote" tests.
-   */
-  void startFlushing() {
-    LOG.info("Triggering HTracedRESTReceiver flush.");
-    lock.lock();
-    try {
-      mustStartFlush = true;
-      cond.signal();
-    } finally {
-      lock.unlock();
-    }
-  }
-
-  private static long WARN_TIMEOUT_MS = 300000;
-
-  @Override
-  public void receiveSpan(Span span) {
-    boolean added = false;
-    lock.lock();
-    try {
-      if (shutdown) {
-        LOG.trace("receiveSpan(span=" + span + "): HTracedRESTReceiver " +
-            "is already shut down.");
-        return;
-      }
-      if (spans.size() < capacity) {
-        spans.add(span);
-        added = true;
-        if (spans.size() >= maxToSendAtATime) {
-          cond.signal();
-        }
-      } else {
-        cond.signal();
-      }
-    } finally {
-      lock.unlock();
-    }
-    if (!added) {
-      long now = TimeUnit.MILLISECONDS.convert(System.nanoTime(),
-          TimeUnit.NANOSECONDS);
-      long last = lastAtCapacityWarningLog.get();
-      if (now - last > WARN_TIMEOUT_MS) {
-        // Only log every 5 minutes. Any more than this for a guest process
-        // is obnoxious.
-        if (lastAtCapacityWarningLog.compareAndSet(last, now)) {
-          // If the atomic-compare-and-set succeeds, we should log.  Otherwise,
-          // we should assume another thread already logged and bumped up the
-          // value of lastAtCapacityWarning sometime between our get and the
-          // "if" statement.
-          LOG.warn("There are too many HTrace spans to buffer!  We have " +
-              "already buffered " + capacity + " spans.  Dropping spans.");
-        }
-      }
-    }
-  }
-}
diff --git a/htrace-htraced/src/main/java/org/apache/htrace/impl/HTracedSpanReceiver.java b/htrace-htraced/src/main/java/org/apache/htrace/impl/HTracedSpanReceiver.java
new file mode 100644
index 0000000..f5f493c
--- /dev/null
+++ b/htrace-htraced/src/main/java/org/apache/htrace/impl/HTracedSpanReceiver.java
@@ -0,0 +1,333 @@
+/**
+ * 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.htrace.impl;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.apache.htrace.core.HTraceConfiguration;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.htrace.core.Span;
+import org.apache.htrace.core.SpanReceiver;
+
+/**
+ * The SpanReceiver which sends spans to htraced.
+ *
+ * HTracedSpanReceiver sends trace spans out to the htraced daemon, where they
+ * are stored and indexed.  It supports two forms of RPC: a JSON/HTTP form, and
+ * an HRPC/msgpack form.  We will use the msgpack form when
+ * htraced.receiver.packed is set to true.
+ *
+ * HTraced buffers are several megabytes in size, and we reuse them to avoid
+ * creating extra garbage on the heap.  They are flushed whenever a timeout
+ * elapses, or when they get more than a configurable percent full.  We allocate
+ * two buffers so that we can continue filling one buffer while the other is
+ * being sent over the wire.  The buffers store serialized spans.  This is
+ * better than storing references to span objects because it minimzes the amount
+ * of pointers we have to follow during a GC.  Buffers are managed by instances
+ * of BufferManager.
+ */
+public class HTracedSpanReceiver extends SpanReceiver {
+  private static final Log LOG = LogFactory.getLog(HTracedSpanReceiver.class);
+
+  private final static int MAX_CLOSING_WAIT_MS = 120000;
+
+  private final FaultInjector faultInjector;
+
+  private final Conf conf;
+
+  private final ReentrantLock lock = new ReentrantLock();
+
+  private final Condition wakePostSpansThread = lock.newCondition();
+
+  private final BufferManager bufferManager[] = new BufferManager[2];
+
+  private final RateLimitedLogger flushErrorLog;
+
+  private final RateLimitedLogger spanDropLog;
+
+  private final PostSpansThread thread;
+
+  private boolean shutdown = false;
+
+  private int activeBuf = 0;
+
+  private int flushingBuf = -1;
+
+  private long lastBufferClearedTimeMs = 0;
+
+  static class FaultInjector {
+    static FaultInjector NO_OP = new FaultInjector();
+    public void handleContentLengthTrigger(int len) { }
+    public void handleThreadStart() throws Exception { }
+    public void handleFlush() throws IOException { }
+  }
+
+  public HTracedSpanReceiver(HTraceConfiguration c) throws Exception {
+    this(c, FaultInjector.NO_OP);
+  }
+
+  HTracedSpanReceiver(HTraceConfiguration c,
+      FaultInjector faultInjector) throws Exception {
+    this.faultInjector = faultInjector;
+    this.conf = new Conf(c);
+    if (this.conf.packed) {
+      for (int i = 0; i < bufferManager.length; i++) {
+        bufferManager[i] = new PackedBufferManager(conf);
+      }
+    } else {
+      for (int i = 0; i < bufferManager.length; i++) {
+        bufferManager[i] = new RestBufferManager(conf);
+      }
+    }
+    this.flushErrorLog = new RateLimitedLogger(LOG, conf.errorLogPeriodMs);
+    this.spanDropLog = new RateLimitedLogger(LOG, conf.errorLogPeriodMs);
+    this.thread = new PostSpansThread();
+    LOG.debug("Created new HTracedSpanReceiver with " + conf.toString());
+  }
+
+  @Override
+  public void receiveSpan(Span span) {
+    long startTimeMs = 0;
+    int numTries = 1;
+    while (true) {
+      lock.lock();
+      try {
+        if (shutdown) {
+          LOG.info("Unable to add span because HTracedSpanReceiver is shutting down.");
+          return;
+        }
+        Throwable exc = null;
+        try {
+          bufferManager[activeBuf].writeSpan(span);
+          int contentLength = bufferManager[activeBuf].contentLength();
+          if (contentLength > conf.triggerSize) {
+            if (LOG.isDebugEnabled()) {
+              LOG.debug("Triggering buffer #" + activeBuf + " flush because" +
+                  " buffer contains " + contentLength + " bytes, and " +
+                  "triggerSize is " + conf.triggerSize);
+            }
+            faultInjector.handleContentLengthTrigger(contentLength);
+            wakePostSpansThread.signal();
+          }
+          return;
+        } catch (Exception e) {
+          exc = e;
+        } catch (Error e) {
+          exc = e;
+        }
+        if (startTimeMs == 0) {
+          startTimeMs = TimeUtil.nowMs();
+        }
+        long deltaMs = TimeUtil.deltaMs(startTimeMs, TimeUtil.nowMs());
+        if (deltaMs > conf.spanDropTimeoutMs) {
+          spanDropLog.error("Dropping a span after unsuccessfully " +
+              "attempting to add it for " + deltaMs + " ms.  There is not " +
+              "enough buffer space. Please increase " + Conf.BUFFER_SIZE_KEY +
+              " or decrease the rate of spans being generated.");
+          return;
+        } else if (LOG.isDebugEnabled()) {
+          LOG.debug("Unable to write span to buffer #" + activeBuf +
+              " after " + numTries + " attempt(s) and " + deltaMs + " ms" +
+              ".  Buffer already has " +
+                  bufferManager[activeBuf].getNumberOfSpans() + " spans.",
+              exc);
+        }
+        numTries++;
+      } finally {
+        lock.unlock();
+      }
+      try {
+        Thread.sleep(conf.spanDropTimeoutMs / 3);
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+      }
+    }
+  }
+
+  @Override
+  public void close() {
+    lock.lock();
+    try {
+      shutdown = true;
+      wakePostSpansThread.signal();
+    } finally {
+      lock.unlock();
+    }
+    try {
+      thread.join(MAX_CLOSING_WAIT_MS);
+    } catch (InterruptedException e) {
+      LOG.error("HTracedSpanReceiver#close was interrupted", e);
+      Thread.currentThread().interrupt();
+    }
+  }
+
+  private class PostSpansThread extends Thread {
+    PostSpansThread() {
+      this.setDaemon(true);
+      this.setName("PostSpans");
+      this.start();
+    }
+
+    private boolean shouldWaitForCond(long timeSinceLastClearedMs) {
+      if (shutdown) {
+        // If we're shutting down, don't wait around.
+        LOG.trace("Should not wait for cond because we're shutting down.");
+        return false;
+      }
+      int contentLength = bufferManager[activeBuf].contentLength();
+      if (contentLength == 0) {
+        // If there is nothing in the buffer, there is nothing to do.
+        if (LOG.isTraceEnabled()) {
+          LOG.trace("Should wait for cond because we have no data buffered " +
+              "in bufferManager " + activeBuf);
+        }
+        lastBufferClearedTimeMs = TimeUtil.nowMs();
+        return true;
+      } else if (contentLength >= conf.triggerSize) {
+        // If the active buffer is filling up, start flushing.
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Should not wait for cond because we have more than " +
+              conf.triggerSize + " bytes buffered in bufferManager " +
+              activeBuf);
+        }
+        return false;
+      }
+      if (timeSinceLastClearedMs > conf.maxFlushIntervalMs) {
+        // If we have let the spans sit in the buffer for too long,
+        // start flushing.
+        if (LOG.isTraceEnabled()) {
+          LOG.trace("Should not wait for cond because it has been " +
+              timeSinceLastClearedMs + " ms since our last flush, and we " +
+              "are overdue for another because maxFlushIntervalMs is " +
+              conf.maxFlushIntervalMs);
+        }
+        return false;
+      }
+      LOG.trace("Should wait for cond.");
+      return true;
+    }
+
+    @Override
+    public void run() {
+      try {
+        faultInjector.handleThreadStart();
+        LOG.debug("Starting HTracedSpanReceiver thread for " +
+            conf.endpointStr);
+        BufferManager flushBufManager = null;
+        while (true) {
+          lock.lock();
+          flushingBuf = -1;
+          try {
+            while (true) {
+              long timeSinceLastClearedMs = TimeUtil.
+                deltaMs(lastBufferClearedTimeMs, TimeUtil.nowMs());
+              if (!shouldWaitForCond(timeSinceLastClearedMs)) {
+                break;
+              }
+              long waitMs = conf.maxFlushIntervalMs -
+                  Math.min(conf.maxFlushIntervalMs, TimeUtil.
+                      deltaMs(TimeUtil.nowMs(), lastBufferClearedTimeMs));
+              if (LOG.isTraceEnabled()) {
+                LOG.trace("Waiting on wakePostSpansThread for " + waitMs +
+                    " ms.");
+              }
+              try {
+                wakePostSpansThread.await(waitMs, TimeUnit.MILLISECONDS);
+              } catch (InterruptedException e) {
+                LOG.info("HTraceSpanReceiver thread was interrupted.", e);
+                throw e;
+              }
+            }
+            if (shutdown && (bufferManager[activeBuf].contentLength() == 0)) {
+              LOG.debug("PostSpansThread shutting down.");
+              return;
+            }
+            flushingBuf = activeBuf;
+            flushBufManager = bufferManager[flushingBuf];
+            activeBuf = (activeBuf == 1) ? 0 : 1;
+          } finally {
+            lock.unlock();
+          }
+          doFlush(flushBufManager);
+          flushBufManager.clear();
+          lastBufferClearedTimeMs = TimeUtil.nowMs();
+          if (LOG.isTraceEnabled()) {
+            LOG.trace("setting lastBufferClearedTimeMs to " + lastBufferClearedTimeMs);
+          }
+        }
+      } catch (Throwable e) {
+        LOG.error("PostSpansThread exiting on unexpected exception", e);
+      } finally {
+        for (int i = 0; i < bufferManager.length; i++) {
+          bufferManager[i].close();
+        }
+      }
+    }
+
+    private void doFlush(BufferManager flushBufManager)
+        throws InterruptedException {
+      try {
+        flushBufManager.prepare();
+      } catch (IOException e) {
+        LOG.error("Failed to prepare buffer containing " +
+            flushBufManager.getNumberOfSpans() + " spans for " +
+            "sending to " + conf.endpointStr + " Discarding " +
+            "all spans.", e);
+        return;
+      }
+      int flushTries = 0;
+      while (true) {
+        Throwable exc;
+        try {
+          faultInjector.handleFlush();
+          flushBufManager.flush();
+          exc = null;
+        } catch (RuntimeException e) {
+          exc = e;
+        } catch (Exception e) {
+          exc = e;
+        }
+        if (exc == null) {
+          return;
+        }
+        int numSpans = flushBufManager.getNumberOfSpans();
+        String excMessage = "Failed to flush " + numSpans  + " htrace " +
+            "spans to " + conf.endpointStr + " on try " + (flushTries + 1);
+        if (flushTries >= conf.flushRetryDelays.length) {
+          excMessage += ".  Discarding all spans.";
+        }
+        if (LOG.isDebugEnabled()) {
+          LOG.error(excMessage, exc);
+        } else {
+          flushErrorLog.error(excMessage, exc);
+        }
+        if (flushTries >= conf.flushRetryDelays.length) {
+          return;
+        }
+        int delayMs = conf.flushRetryDelays[flushTries];
+        Thread.sleep(delayMs);
+        flushTries++;
+      }
+    }
+  }
+}
diff --git a/htrace-htraced/src/main/java/org/apache/htrace/impl/PackedBuffer.java b/htrace-htraced/src/main/java/org/apache/htrace/impl/PackedBuffer.java
new file mode 100644
index 0000000..f867ad7
--- /dev/null
+++ b/htrace-htraced/src/main/java/org/apache/htrace/impl/PackedBuffer.java
@@ -0,0 +1,449 @@
+/**
+ * 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.htrace.impl;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.htrace.core.MilliSpan;
+import org.apache.htrace.core.TimelineAnnotation;
+import org.msgpack.core.MessagePack;
+import org.msgpack.core.MessagePacker;
+import org.msgpack.core.MessageUnpacker;
+import org.msgpack.core.buffer.MessageBuffer;
+import org.msgpack.core.buffer.MessageBufferOutput;
+
+import org.apache.htrace.core.Span;
+import org.apache.htrace.core.SpanId;
+
+/**
+ * A ByteBuffer which we are writing msgpack data to.
+ */
+class PackedBuffer {
+  /**
+   * A MessageBufferOutput that simply outputs to a ByteBuffer.
+   */
+  private class PackedBufferOutput implements MessageBufferOutput {
+    private MessageBuffer savedBuffer;
+
+    PackedBufferOutput() {
+    }
+
+    @Override
+    public MessageBuffer next(int bufferSize) throws IOException {
+      if (savedBuffer == null || savedBuffer.size() != bufferSize) {
+        savedBuffer = MessageBuffer.newBuffer(bufferSize);
+      }
+      MessageBuffer buffer = savedBuffer;
+      savedBuffer = null;
+      return buffer;
+    }
+
+    @Override
+    public void flush(MessageBuffer buffer) throws IOException {
+      ByteBuffer b = buffer.toByteBuffer();
+      bb.put(b);
+      savedBuffer = buffer;
+    }
+
+    @Override
+    public void close() throws IOException {
+      // no-op
+    }
+  }
+
+  private static final Log LOG = LogFactory.getLog(PackedBuffer.class);
+  private static final Charset UTF8 = StandardCharsets.UTF_8;
+  private static final byte SPANS[] = "Spans".getBytes(UTF8);
+  private static final byte DEFAULT_PID[] = "DefaultPid".getBytes(UTF8);
+  private static final byte A[] = "a".getBytes(UTF8);
+  private static final byte B[] = "b".getBytes(UTF8);
+  private static final byte E[] = "e".getBytes(UTF8);
+  private static final byte D[] = "d".getBytes(UTF8);
+  private static final byte R[] = "r".getBytes(UTF8);
+  private static final byte P[] = "p".getBytes(UTF8);
+  private static final byte N[] = "n".getBytes(UTF8);
+  private static final byte T[] = "t".getBytes(UTF8);
+  private static final byte M[] = "m".getBytes(UTF8);
+  private static final int HRPC_MAGIC = 0x43525448;
+  static final int HRPC_REQ_FRAME_LENGTH = 20;
+  static final int HRPC_RESP_FRAME_LENGTH = 20;
+  static final int MAX_HRPC_ERROR_LENGTH = 4 * 1024 * 1024;
+  static final int MAX_HRPC_BODY_LENGTH = 64 * 1024 * 1024;
+  private static final int SPAN_ID_BYTE_LENGTH = 16;
+  static final MessagePack.Config MSGPACK_CONF =
+      new MessagePack.ConfigBuilder()
+        .readBinaryAsString(false)
+        .readStringAsBinary(false)
+        .build();
+  /**
+   * The array which we are filling.
+   */
+  final ByteBuffer bb;
+
+  /**
+   * Used to tell the MessagePacker to output to our array.
+   */
+  final PackedBufferOutput out;
+
+  /**
+   * A temporary buffer for serializing span ids and other things.
+   */
+  final byte[] temp;
+
+  /**
+   * Generates msgpack output.
+   */
+  final MessagePacker packer;
+
+  /**
+   * Create a new PackedBuffer.
+   *
+   * @param bb        The ByteBuffer to use to create the packed buffer.
+   */
+  PackedBuffer(ByteBuffer bb) {
+    this.bb = bb;
+    this.out = new PackedBufferOutput();
+    this.temp = new byte[SPAN_ID_BYTE_LENGTH];
+    this.packer = new MessagePacker(out, MSGPACK_CONF);
+  }
+
+  /**
+   * Write the fixed-length request frame which starts packed RPC messages.
+   */
+  static void writeReqFrame(ByteBuffer bb, int methodId, long seq, int length)
+      throws IOException {
+    int oldPos = bb.position();
+    boolean success = false;
+    try {
+      bb.order(ByteOrder.LITTLE_ENDIAN);
+      bb.putInt(HRPC_MAGIC);
+      bb.putInt(methodId);
+      bb.putLong(seq);
+      bb.putInt(length);
+      success = true;
+    } finally {
+      if (!success) {
+        bb.position(oldPos);
+      }
+    }
+  }
+
+  /**
+   * Write an 8-byte value to a byte array as little-endian.
+   */
+  private static void longToBigEndian(byte b[], int pos, long val) {
+    b[pos + 0] =(byte) ((val >> 56) & 0xff);
+    b[pos + 1] =(byte) ((val >> 48) & 0xff);
+    b[pos + 2] =(byte) ((val >> 40) & 0xff);
+    b[pos + 3] =(byte) ((val >> 32) & 0xff);
+    b[pos + 4] =(byte) ((val >> 24) & 0xff);
+    b[pos + 5] =(byte) ((val >> 16) & 0xff);
+    b[pos + 6] =(byte) ((val >>  8) & 0xff);
+    b[pos + 7] =(byte) ((val >>  0) & 0xff);
+  }
+
+  private void writeSpanId(SpanId spanId) throws IOException {
+    longToBigEndian(temp, 0, spanId.getHigh());
+    longToBigEndian(temp, 8, spanId.getLow());
+    packer.packBinaryHeader(SPAN_ID_BYTE_LENGTH);
+    packer.writePayload(temp, 0, SPAN_ID_BYTE_LENGTH);
+  }
+
+  /**
+   * Serialize a span to the given OutputStream.
+   */
+  void writeSpan(Span span) throws IOException {
+    boolean success = false;
+    int oldPos = bb.position();
+    try {
+      int mapSize = 0;
+      if (span.getSpanId().isValid()) {
+        mapSize++;
+      }
+      if (span.getStartTimeMillis() != 0) {
+        mapSize++;
+      }
+      if (span.getStopTimeMillis() != 0) {
+        mapSize++;
+      }
+      if (!span.getDescription().isEmpty()) {
+        mapSize++;
+      }
+      if (!span.getTracerId().isEmpty()) {
+        mapSize++;
+      }
+      if (span.getParents().length > 0) {
+        mapSize++;
+      }
+      if (!span.getKVAnnotations().isEmpty()) {
+        mapSize++;
+      }
+      if (!span.getTimelineAnnotations().isEmpty()) {
+        mapSize++;
+      }
+      packer.packMapHeader(mapSize);
+      if (span.getSpanId().isValid()) {
+        packer.packRawStringHeader(1);
+        packer.writePayload(A);
+        writeSpanId(span.getSpanId());
+      }
+      if (span.getStartTimeMillis() != 0) {
+        packer.packRawStringHeader(1);
+        packer.writePayload(B);
+        packer.packLong(span.getStartTimeMillis());
+      }
+      if (span.getStopTimeMillis() != 0) {
+        packer.packRawStringHeader(1);
+        packer.writePayload(E);
+        packer.packLong(span.getStopTimeMillis());
+      }
+      if (!span.getDescription().isEmpty()) {
+        packer.packRawStringHeader(1);
+        packer.writePayload(D);
+        packer.packString(span.getDescription());
+      }
+      if (!span.getTracerId().isEmpty()) {
+        packer.packRawStringHeader(1);
+        packer.writePayload(R);
+        packer.packString(span.getTracerId());
+      }
+      if (span.getParents().length > 0) {
+        packer.packRawStringHeader(1);
+        packer.writePayload(P);
+        packer.packArrayHeader(span.getParents().length);
+        for (int i = 0; i < span.getParents().length; i++) {
+          writeSpanId(span.getParents()[i]);
+        }
+      }
+      if (!span.getKVAnnotations().isEmpty()) {
+        packer.packRawStringHeader(1);
+        packer.writePayload(N);
+        Map<String, String> map = span.getKVAnnotations();
+        packer.packMapHeader(map.size());
+        for (Map.Entry<String, String> entry : map.entrySet()) {
+          packer.packString(entry.getKey());
+          packer.packString(entry.getValue());
+        }
+      }
+      if (!span.getTimelineAnnotations().isEmpty()) {
+        packer.packRawStringHeader(1);
+        packer.writePayload(T);
+        List<TimelineAnnotation> list = span.getTimelineAnnotations();
+        packer.packArrayHeader(list.size());
+        for (TimelineAnnotation annotation : list) {
+          packer.packMapHeader(2);
+          packer.packRawStringHeader(1);
+          packer.writePayload(T);
+          packer.packLong(annotation.getTime());
+          packer.packRawStringHeader(1);
+          packer.writePayload(M);
+          packer.packString(annotation.getMessage());
+        }
+      }
+      packer.flush();
+      success = true;
+    } finally {
+      if (!success) {
+        // If we failed earlier, restore the old position.
+        // This is so that if we run out of space, we don't add a
+        // partial span to the buffer.
+        bb.position(oldPos);
+      }
+    }
+  }
+
+  static SpanId readSpanId(MessageUnpacker unpacker) throws IOException {
+    int alen = unpacker.unpackBinaryHeader();
+    if (alen != SPAN_ID_BYTE_LENGTH) {
+      throw new IOException("Invalid length given for spanID array.  " +
+          "Expected " + SPAN_ID_BYTE_LENGTH + "; got " + alen);
+    }
+    byte[] payload = new byte[SPAN_ID_BYTE_LENGTH];
+    unpacker.readPayload(payload);
+    return new SpanId(
+        ((payload[ 7] & 0xffL) <<  0) |
+        ((payload[ 6] & 0xffL) <<  8) |
+        ((payload[ 5] & 0xffL) << 16) |
+        ((payload[ 4] & 0xffL) << 24) |
+        ((payload[ 3] & 0xffL) << 32) |
+        ((payload[ 2] & 0xffL) << 40) |
+        ((payload[ 1] & 0xffL) << 48) |
+        ((payload[ 0] & 0xffL) << 56),
+        ((payload[15] & 0xffL) <<  0) |
+        ((payload[14] & 0xffL) <<  8) |
+        ((payload[13] & 0xffL) << 16) |
+        ((payload[12] & 0xffL) << 24) |
+        ((payload[11] & 0xffL) << 32) |
+        ((payload[10] & 0xffL) << 40) |
+        ((payload[ 9] & 0xffL) << 48) |
+        ((payload[ 8] & 0xffL) << 56)
+      );
+  }
+
+  /**
+   * Read a span.  Used in unit tests.  Not optimized.
+   */
+  static Span readSpan(MessageUnpacker unpacker) throws IOException {
+    int numEntries = unpacker.unpackMapHeader();
+    MilliSpan.Builder builder = new MilliSpan.Builder();
+    while (--numEntries >= 0) {
+      String key = unpacker.unpackString();
+      if (key.length() != 1) {
+        throw new IOException("Unknown key " + key);
+      }
+      switch (key.charAt(0)) {
+        case 'a':
+          builder.spanId(readSpanId(unpacker));
+          break;
+        case 'b':
+          builder.begin(unpacker.unpackLong());
+          break;
+        case 'e':
+          builder.end(unpacker.unpackLong());
+          break;
+        case 'd':
+          builder.description(unpacker.unpackString());
+          break;
+        case 'r':
+          builder.tracerId(unpacker.unpackString());
+          break;
+        case 'p':
+          int numParents = unpacker.unpackArrayHeader();
+          SpanId[] parents = new SpanId[numParents];
+          for (int i = 0; i < numParents; i++) {
+            parents[i] = readSpanId(unpacker);
+          }
+          builder.parents(parents);
+          break;
+        case 'n':
+          int mapEntries = unpacker.unpackMapHeader();
+          HashMap<String, String> entries =
+              new HashMap<String, String>(mapEntries);
+          for (int i = 0; i < mapEntries; i++) {
+            String k = unpacker.unpackString();
+            String v = unpacker.unpackString();
+            entries.put(k, v);
+          }
+          builder.traceInfo(entries);
+          break;
+        case 't':
+          int listEntries = unpacker.unpackArrayHeader();
+          ArrayList<TimelineAnnotation> list =
+              new ArrayList<TimelineAnnotation>(listEntries);
+          for (int i = 0; i < listEntries; i++) {
+            int timelineObjectSize = unpacker.unpackMapHeader();
+            long time = 0;
+            String msg = "";
+            for (int j = 0; j < timelineObjectSize; j++) {
+              String tlKey = unpacker.unpackString();
+              if (tlKey.length() != 1) {
+                throw new IOException("Unknown timeline map key " + tlKey);
+              }
+              switch (tlKey.charAt(0)) {
+                case 't':
+                  time = unpacker.unpackLong();
+                  break;
+                case 'm':
+                  msg = unpacker.unpackString();
+                  break;
+                default:
+                  throw new IOException("Unknown timeline map key " + tlKey);
+              }
+            }
+            list.add(new TimelineAnnotation(time, msg));
+          }
+          builder.timeline(list);
+          break;
+        default:
+          throw new IOException("Unknown key " + key);
+      }
+    }
+    return builder.build();
+  }
+
+  void beginWriteSpansRequest(String defaultPid, int numSpans)
+      throws IOException {
+    boolean success = false;
+    int oldPos = bb.position();
+    try {
+      int mapSize = 1;
+      if (defaultPid != null) {
+        mapSize++;
+      }
+      packer.packMapHeader(mapSize);
+      if (defaultPid != null) {
+        packer.packRawStringHeader(DEFAULT_PID.length);
+        packer.writePayload(DEFAULT_PID);
+        packer.packString(defaultPid);
+      }
+      packer.packRawStringHeader(SPANS.length);
+      packer.writePayload(SPANS);
+      packer.packArrayHeader(numSpans);
+      packer.flush();
+      success = true;
+    } finally {
+      if (!success) {
+        bb.position(oldPos);
+      }
+    }
+  }
+
+  /**
+   * Get the underlying ByteBuffer.
+   */
+  ByteBuffer getBuffer() {
+    return bb;
+  }
+
+  /**
+   * Reset our position in the array.
+   */
+  void reset() throws IOException {
+    packer.reset(out);
+  }
+
+  void close() {
+    try {
+      packer.close();
+    } catch (IOException e) {
+      LOG.error("Error closing MessagePacker", e);
+    }
+  }
+
+  public String toHexString() {
+    String prefix = "";
+    StringBuilder bld = new StringBuilder();
+    ByteBuffer b = bb.duplicate();
+    b.flip();
+    while (b.hasRemaining()) {
+      bld.append(String.format("%s%02x", prefix, b.get()));
+      prefix = " ";
+    }
+    return bld.toString();
+  }
+}
diff --git a/htrace-htraced/src/main/java/org/apache/htrace/impl/PackedBufferManager.java b/htrace-htraced/src/main/java/org/apache/htrace/impl/PackedBufferManager.java
new file mode 100644
index 0000000..8b59a72
--- /dev/null
+++ b/htrace-htraced/src/main/java/org/apache/htrace/impl/PackedBufferManager.java
@@ -0,0 +1,340 @@
+/**
+ * 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.htrace.impl;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.CharBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.SocketChannel;
+import java.nio.channels.spi.SelectorProvider;
+import java.nio.charset.StandardCharsets;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.apache.htrace.core.Span;
+
+class PackedBufferManager implements BufferManager {
+  private static final Log LOG = LogFactory.getLog(PackedBuffer.class);
+  private static final int MAX_PREQUEL_LENGTH = 2048;
+  private static final int METHOD_ID_WRITE_SPANS = 0x1;
+  private final Conf conf;
+  private final ByteBuffer frameBuffer;
+  private final PackedBuffer prequel;
+  private final PackedBuffer spans;
+  private final Selector selector;
+  private int numSpans;
+
+  PackedBufferManager(Conf conf) throws IOException {
+    this.conf = conf;
+    this.frameBuffer = ByteBuffer.allocate(PackedBuffer.HRPC_REQ_FRAME_LENGTH);
+    this.prequel = new PackedBuffer(ByteBuffer.allocate(MAX_PREQUEL_LENGTH));
+    this.spans = new PackedBuffer(ByteBuffer.allocate(conf.bufferSize));
+    this.selector = SelectorProvider.provider().openSelector();
+    clear();
+  }
+
+  @Override
+  public void writeSpan(Span span) throws IOException {
+    spans.writeSpan(span);
+    numSpans++;
+    if (LOG.isTraceEnabled()) {
+      LOG.trace("wrote " + span.toJson() + " to PackedBuffer for " +
+          conf.endpointStr + ". numSpans = " + numSpans +
+          ", buffer position = " + spans.getBuffer().position());
+    }
+  }
+
+  @Override
+  public int contentLength() {
+    return spans.getBuffer().position();
+  }
+
+  @Override
+  public int getNumberOfSpans() {
+    return numSpans;
+  }
+
+  @Override
+  public void prepare() throws IOException {
+    prequel.beginWriteSpansRequest(null, numSpans);
+    long totalLength =
+        prequel.getBuffer().position() + spans.getBuffer().position();
+    if (totalLength > PackedBuffer.MAX_HRPC_BODY_LENGTH) {
+      throw new IOException("Can't send RPC of " + totalLength + " bytes " +
+          "because it is longer than " + PackedBuffer.MAX_HRPC_BODY_LENGTH);
+    }
+    PackedBuffer.writeReqFrame(frameBuffer,
+        METHOD_ID_WRITE_SPANS, 1, (int)totalLength);
+    frameBuffer.flip();
+    prequel.getBuffer().flip();
+    spans.getBuffer().flip();
+    if (LOG.isTraceEnabled()) {
+      LOG.trace("Preparing to send RPC of length " +
+          (totalLength + PackedBuffer.HRPC_REQ_FRAME_LENGTH) + " to " +
+          conf.endpointStr + ", containing " + numSpans + " spans.");
+    }
+  }
+
+  @Override
+  public void flush() throws IOException {
+    SelectionKey sockKey = null;
+    IOException ioe = null;
+    frameBuffer.position(0);
+    prequel.getBuffer().position(0);
+    spans.getBuffer().position(0);
+    if (LOG.isTraceEnabled()) {
+      LOG.trace("Preparing to flush " + numSpans + " spans to " +
+          conf.endpointStr);
+    }
+    try {
+      sockKey = doConnect();
+      doSend(sockKey, new ByteBuffer[] {
+          frameBuffer, prequel.getBuffer(), spans.getBuffer() });
+      ByteBuffer response = prequel.getBuffer();
+      readAndValidateResponseFrame(sockKey, response,
+          1, METHOD_ID_WRITE_SPANS);
+    } catch (IOException e) {
+      // This LOG message is only at debug level because we also log these
+      // exceptions at error level inside HTracedReceiver.  The logging in
+      // HTracedReceiver is rate-limited to avoid overwhelming the client log
+      // if htraced goes down.  The debug and trace logging is not
+      // rate-limited.
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("Got exception during flush", e);
+      }
+      ioe = e;
+    } finally {
+      if (sockKey != null) {
+        sockKey.cancel();
+        try {
+          SocketChannel sock = (SocketChannel)sockKey.attachment();
+          sock.close();
+        } catch (IOException e) {
+          if (ioe != null) {
+            ioe.addSuppressed(e);
+          }
+        }
+      }
+    }
+    if (ioe != null) {
+      throw ioe;
+    }
+    if (LOG.isTraceEnabled()) {
+      LOG.trace("Successfully flushed " + numSpans + " spans to " +
+          conf.endpointStr);
+    }
+  }
+
+  private long updateRemainingMs(long startMs, long timeoutMs) {
+    long deltaMs = TimeUtil.deltaMs(startMs, TimeUtil.nowMs());
+    if (deltaMs > timeoutMs) {
+      return 0;
+    }
+    return timeoutMs - deltaMs;
+  }
+
+  private SelectionKey doConnect() throws IOException {
+    SocketChannel sock = SocketChannel.open();
+    SelectionKey sockKey = null;
+    boolean success = false;
+    try {
+      if (sock.isBlocking()) {
+        sock.configureBlocking(false);
+      }
+      InetSocketAddress resolvedEndpoint =
+          new InetSocketAddress(conf.endpoint.getHostString(),
+              conf.endpoint.getPort());
+      resolvedEndpoint.getHostName(); // trigger DNS resolution
+      sock.connect(resolvedEndpoint);
+      sockKey = sock.register(selector, SelectionKey.OP_CONNECT, sock);
+      long startMs = TimeUtil.nowMs();
+      long remainingMs = conf.connectTimeoutMs;
+      while (true) {
+        selector.select(remainingMs);
+        for (SelectionKey key : selector.keys()) {
+          if (key.isConnectable()) {
+            SocketChannel s = (SocketChannel)key.attachment();
+            s.finishConnect();
+            if (LOG.isTraceEnabled()) {
+              LOG.trace("Successfully connected to " + conf.endpointStr + ".");
+            }
+            success = true;
+            return sockKey;
+          }
+        }
+        remainingMs = updateRemainingMs(startMs, conf.connectTimeoutMs);
+        if (remainingMs == 0) {
+          throw new IOException("Attempt to connect to " + conf.endpointStr +
+              " timed out after " +  TimeUtil.deltaMs(startMs, TimeUtil.nowMs()) +
+              " ms.");
+        }
+      }
+    } finally {
+      if (!success) {
+        if (sockKey != null) {
+          sockKey.cancel();
+        }
+        sock.close();
+      }
+    }
+  }
+
+  /**
+   * Send the provided ByteBuffer objects.
+   *
+   * We use non-blocking I/O because Java does not provide write timeouts.
+   * Without a write timeout, the socket could get hung and we'd never recover.
+   * We also use the GatheringByteChannel#write method which calls the pread()
+   * system call under the covers.  This ensures that even if TCP_NODELAY is on,
+   * we send the minimal number of packets.
+   */
+  private void doSend(SelectionKey sockKey, ByteBuffer[] bufs)
+        throws IOException {
+    long totalWritten = 0;
+    sockKey.interestOps(SelectionKey.OP_WRITE);
+    SocketChannel sock = (SocketChannel)sockKey.attachment();
+    long startMs = TimeUtil.nowMs();
+    long remainingMs = conf.ioTimeoutMs;
+    while (true) {
+      selector.select(remainingMs);
+      int firstBuf = 0;
+      for (SelectionKey key : selector.selectedKeys()) {
+        if (key.isWritable()) {
+          long written = sock.write(bufs, firstBuf, bufs.length - firstBuf);
+          if (LOG.isTraceEnabled()) {
+            LOG.trace("Sent " + written + " bytes to " + conf.endpointStr);
+          }
+          totalWritten += written;
+        }
+      }
+      while (true) {
+        if (firstBuf == bufs.length) {
+          if (LOG.isTraceEnabled()) {
+            LOG.trace("Finished sending " + totalWritten + " bytes to " +
+                conf.endpointStr);
+          }
+          return;
+        }
+        if (bufs[firstBuf].remaining() > 0) {
+          break;
+        }
+        firstBuf++;
+      }
+      remainingMs = updateRemainingMs(startMs, conf.ioTimeoutMs);
+      if (remainingMs == 0) {
+        throw new IOException("Attempt to write to " + conf.endpointStr +
+            " timed out after " + TimeUtil.deltaMs(startMs, TimeUtil.nowMs()) +
+            " ms.");
+      }
+    }
+  }
+
+  private void doRecv(SelectionKey sockKey, ByteBuffer response)
+      throws IOException {
+    sockKey.interestOps(SelectionKey.OP_READ);
+    SocketChannel sock = (SocketChannel)sockKey.attachment();
+    int totalRead = response.remaining();
+    long startMs = TimeUtil.nowMs();
+    long remainingMs = conf.ioTimeoutMs;
+    while (remainingMs > 0) {
+      selector.select(remainingMs);
+      for (SelectionKey key : selector.selectedKeys()) {
+        if (key.isReadable()) {
+          sock.read(response);
+        }
+      }
+      if (response.remaining() == 0) {
+        if (LOG.isTraceEnabled()) {
+          LOG.trace("Received all " + totalRead + " bytes from " +
+              conf.endpointStr);
+        }
+        return;
+      }
+      remainingMs = updateRemainingMs(startMs, conf.ioTimeoutMs);
+      if (LOG.isTraceEnabled()) {
+        LOG.trace("Received " + (totalRead - response.remaining()) +
+                " out of " + totalRead + " bytes from " + conf.endpointStr);
+      }
+      if (remainingMs == 0) {
+        throw new IOException("Attempt to write to " + conf.endpointStr +
+            " timed out after " + TimeUtil.deltaMs(startMs, TimeUtil.nowMs()) +
+            " ms.");
+      }
+    }
+  }
+
+  private void readAndValidateResponseFrame(SelectionKey sockKey,
+        ByteBuffer buf, long expectedSeq, int expectedMethodId)
+          throws IOException {
+    buf.clear();
+    buf.limit(PackedBuffer.HRPC_RESP_FRAME_LENGTH);
+    doRecv(sockKey, buf);
+    buf.flip();
+    buf.order(ByteOrder.LITTLE_ENDIAN);
+    long seq = buf.getLong();
+    if (seq != expectedSeq) {
+      throw new IOException("Expected sequence number " + expectedSeq +
+          ", but got sequence number " + seq);
+    }
+    int methodId = buf.getInt();
+    if (expectedMethodId != methodId) {
+      throw new IOException("Expected method id " + expectedMethodId +
+          ", but got " + methodId);
+    }
+    int errorLength = buf.getInt();
+    buf.getInt();
+    if ((errorLength < 0) ||
+        (errorLength > PackedBuffer.MAX_HRPC_ERROR_LENGTH)) {
+      throw new IOException("Got server error with invalid length " +
+          errorLength);
+    } else if (errorLength > 0) {
+      buf.clear();
+      buf.limit(errorLength);
+      doRecv(sockKey, buf);
+      buf.flip();
+      CharBuffer charBuf = StandardCharsets.UTF_8.decode(buf);
+      String serverErrorStr = charBuf.toString();
+      throw new IOException("Got server error " + serverErrorStr);
+    }
+  }
+
+  @Override
+  public void clear() {
+    frameBuffer.clear();
+    prequel.getBuffer().clear();
+    spans.getBuffer().clear();
+    numSpans = 0;
+  }
+
+  @Override
+  public void close() {
+    clear();
+    prequel.close();
+    spans.close();
+    try {
+      selector.close();
+    } catch (IOException e) {
+      LOG.warn("Error closing selector", e);
+    }
+  }
+}
diff --git a/htrace-htraced/src/main/java/org/apache/htrace/impl/RateLimitedLogger.java b/htrace-htraced/src/main/java/org/apache/htrace/impl/RateLimitedLogger.java
new file mode 100644
index 0000000..ac42ee8
--- /dev/null
+++ b/htrace-htraced/src/main/java/org/apache/htrace/impl/RateLimitedLogger.java
@@ -0,0 +1,72 @@
+/**
+ * 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.htrace.impl;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.logging.Log;
+
+/**
+ * A logger which rate-limits its logging to a configurable level.
+ */
+class RateLimitedLogger {
+  private final Log log;
+  private final long timeoutMs;
+  private long lastLogTimeMs;
+
+  public RateLimitedLogger(Log log, long timeoutMs) {
+    this.log = log;
+    this.timeoutMs = timeoutMs;
+    synchronized (this) {
+      this.lastLogTimeMs = 0L;
+    }
+  }
+
+  public void warn(String what) {
+    long now = TimeUnit.MILLISECONDS.convert(System.nanoTime(),
+        TimeUnit.NANOSECONDS);
+    synchronized (this) {
+      if (now >= lastLogTimeMs + timeoutMs) {
+        log.warn(what);
+        lastLogTimeMs = now;
+      }
+    }
+  }
+
+  public void error(String what) {
+    long now = TimeUnit.MILLISECONDS.convert(System.nanoTime(),
+        TimeUnit.NANOSECONDS);
+    synchronized (this) {
+      if (now >= lastLogTimeMs + timeoutMs) {
+        log.error(what);
+        lastLogTimeMs = now;
+      }
+    }
+  }
+
+  public void error(String what, Throwable e) {
+    long now = TimeUnit.MILLISECONDS.convert(System.nanoTime(),
+        TimeUnit.NANOSECONDS);
+    synchronized (this) {
+      if (now >= lastLogTimeMs + timeoutMs) {
+        log.error(what, e);
+        lastLogTimeMs = now;
+      }
+    }
+  }
+}
diff --git a/htrace-htraced/src/main/java/org/apache/htrace/impl/RestBufferManager.java b/htrace-htraced/src/main/java/org/apache/htrace/impl/RestBufferManager.java
new file mode 100644
index 0000000..2e1aa70
--- /dev/null
+++ b/htrace-htraced/src/main/java/org/apache/htrace/impl/RestBufferManager.java
@@ -0,0 +1,225 @@
+/**
+ * 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.htrace.impl;
+
+import java.io.IOException;
+import java.net.URL;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.apache.htrace.core.Span;
+import org.eclipse.jetty.client.HttpClient;
+import org.eclipse.jetty.client.api.ContentProvider;
+import org.eclipse.jetty.client.api.ContentResponse;
+import org.eclipse.jetty.client.api.Request;
+import org.eclipse.jetty.http.HttpField;
+import org.eclipse.jetty.http.HttpHeader;
+import org.eclipse.jetty.http.HttpMethod;
+import org.eclipse.jetty.http.HttpStatus;
+
+class RestBufferManager implements BufferManager {
+  private static final Log LOG = LogFactory.getLog(RestBufferManager.class);
+  private static ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+  private static ObjectWriter JSON_WRITER = OBJECT_MAPPER.writer();
+  private static final Charset UTF8 = Charset.forName("UTF-8");
+  private static final byte COMMA_BYTE = (byte)0x2c;
+  private static final int MAX_PREQUEL_LENGTH = 512;
+  private static final int MAX_EPILOGUE_LENGTH = 32;
+  private final Conf conf;
+  private final HttpClient httpClient;
+  private final String urlString;
+  private final ByteBuffer prequel;
+  private final ByteBuffer spans;
+  private final ByteBuffer epilogue;
+  private int numSpans;
+
+  private static class RestBufferManagerContentProvider
+      implements ContentProvider {
+    private final ByteBuffer[] bufs;
+
+    private class ByteBufferIterator implements Iterator<ByteBuffer> {
+      private int bufIdx = -1;
+
+      @Override
+      public boolean hasNext() {
+        return (bufIdx + 1) < bufs.length;
+      }
+
+      @Override
+      public ByteBuffer next() {
+        if ((bufIdx + 1) >= bufs.length) {
+          throw new NoSuchElementException();
+        }
+        bufIdx++;
+        return bufs[bufIdx];
+      }
+
+      @Override
+      public void remove() {
+        throw new UnsupportedOperationException();
+      }
+    }
+
+    RestBufferManagerContentProvider(ByteBuffer[] bufs) {
+      this.bufs = bufs;
+    }
+
+    @Override
+    public long getLength() {
+      long total = 0;
+      for (int i = 0; i < bufs.length; i++) {
+        total += bufs[i].remaining();
+      }
+      return total;
+    }
+
+    @Override
+    public Iterator<ByteBuffer> iterator() {
+      return new ByteBufferIterator();
+    }
+  }
+
+  /**
+   * Create an HttpClient instance.
+   *
+   * @param connTimeout         The timeout to use for connecting.
+   * @param idleTimeout         The idle timeout to use.
+   */
+  static HttpClient createHttpClient(long connTimeout, long idleTimeout) {
+    HttpClient httpClient = new HttpClient();
+    httpClient.setUserAgentField(
+        new HttpField(HttpHeader.USER_AGENT, "HTracedSpanReceiver"));
+    httpClient.setConnectTimeout(connTimeout);
+    httpClient.setIdleTimeout(idleTimeout);
+    return httpClient;
+  }
+
+  RestBufferManager(Conf conf) throws Exception {
+    this.conf = conf;
+    this.httpClient =
+        createHttpClient(conf.connectTimeoutMs, conf.idleTimeoutMs);
+    this.urlString = new URL("http", conf.endpoint.getHostName(),
+        conf.endpoint.getPort(), "/writeSpans").toString();
+    this.prequel = ByteBuffer.allocate(MAX_PREQUEL_LENGTH);
+    this.spans = ByteBuffer.allocate(conf.bufferSize);
+    this.epilogue = ByteBuffer.allocate(MAX_EPILOGUE_LENGTH);
+    clear();
+    this.httpClient.start();
+  }
+
+  @Override
+  public void writeSpan(Span span) throws IOException {
+    byte[] spanJsonBytes = JSON_WRITER.writeValueAsBytes(span);
+    if ((spans.capacity() - spans.position()) < (spanJsonBytes.length + 1)) {
+      // Make sure we have enough space for the span JSON and a comma.
+      throw new IOException("Not enough space remaining in span buffer.");
+    }
+    spans.put(COMMA_BYTE);
+    spans.put(spanJsonBytes);
+    numSpans++;
+  }
+
+  @Override
+  public int contentLength() {
+    return Math.max(spans.position() - 1, 0);
+  }
+
+  @Override
+  public int getNumberOfSpans() {
+    return numSpans;
+  }
+
+  @Override
+  public void prepare() throws IOException {
+    String prequelString = "{\"Spans\":[";
+    prequel.put(prequelString.getBytes(UTF8));
+    prequel.flip();
+
+    spans.flip();
+
+    String epilogueString = "]}";
+    epilogue.put(epilogueString.toString().getBytes(UTF8));
+    epilogue.flip();
+
+    if (LOG.isTraceEnabled()) {
+      LOG.trace("Preparing to send " + contentLength() + " bytes of span " +
+          "data to " + conf.endpointStr + ", containing " + numSpans +
+          " spans.");
+    }
+  }
+
+  @Override
+  public void flush() throws IOException {
+    // Position the buffers at the beginning.
+    prequel.position(0);
+    spans.position(spans.limit() == 0 ? 0 : 1); // Skip the first comma
+    epilogue.position(0);
+
+    RestBufferManagerContentProvider contentProvider =
+        new RestBufferManagerContentProvider(
+            new ByteBuffer[] { prequel, spans, epilogue });
+    long rpcLength = contentProvider.getLength();
+    try {
+      Request request = httpClient.
+          newRequest(urlString).method(HttpMethod.POST);
+      request.header(HttpHeader.CONTENT_TYPE, "application/json");
+      request.content(contentProvider);
+      ContentResponse response = request.send();
+      if (response.getStatus() != HttpStatus.OK_200) {
+        throw new IOException("Got back error response " +
+            response.getStatus() + " from " + conf.endpointStr + "; " +
+            response.getContentAsString());
+      }
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("Sent WriteSpansReq of length " + rpcLength + " to " + conf.endpointStr);
+      }
+    } catch (InterruptedException e) {
+      throw new IOException("Interrupted while sending spans via REST", e);
+    } catch (TimeoutException e) {
+      throw new IOException("Timed out sending spans via REST", e);
+    } catch (ExecutionException e) {
+      throw new IOException("Execution exception sending spans via REST", e);
+    }
+  }
+
+  @Override
+  public void clear() {
+    prequel.clear();
+    spans.clear();
+    epilogue.clear();
+    numSpans = 0;
+  }
+
+  @Override
+  public void close() {
+    try {
+      httpClient.stop();
+    } catch (Exception e) {
+      LOG.error("Error stopping HTracedReceiver httpClient", e);
+    }
+  }
+}
diff --git a/htrace-htraced/src/main/java/org/apache/htrace/impl/TimeUtil.java b/htrace-htraced/src/main/java/org/apache/htrace/impl/TimeUtil.java
new file mode 100644
index 0000000..7361c97
--- /dev/null
+++ b/htrace-htraced/src/main/java/org/apache/htrace/impl/TimeUtil.java
@@ -0,0 +1,78 @@
+/**
+ * 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.htrace.impl;
+
+import java.math.BigInteger;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Utilities for dealing with monotonic time.
+ */
+class TimeUtil {
+  /**
+   * Returns the current monotonic time in milliseconds.
+   */
+  static long nowMs() {
+    return TimeUnit.MILLISECONDS.convert(
+        System.nanoTime(), TimeUnit.NANOSECONDS);
+  }
+
+  /**
+   * Get the approximate delta between two monotonic times.
+   *
+   * This function makes the following assumptions:
+   * 1. We read startMs from the monotonic clock prior to endMs.
+   * 2. The two times are not more than 100 years or so apart.
+   *
+   * With these two assumptions in hand, we can smooth over some of the
+   * unpleasant features of the monotonic clock:
+   * 1. It can return either positive or negative values.
+   * 2. When the number of nanoseconds reaches Long.MAX_VALUE it wraps around
+   * to Long.MIN_VALUE.
+   * 3. On some messed up systems it has been known to jump backwards every
+   * now and then.  Oops.  CPU core synchronization mumble mumble.
+   *
+   * @param startMs  The start time.
+   * @param endMs    The end time.
+   * @return         The delta between the two times.
+   */
+  static long deltaMs(long startMs, long endMs) {
+    BigInteger startNs = BigInteger.valueOf(TimeUnit.NANOSECONDS.
+        convert(startMs, TimeUnit.MILLISECONDS));
+    BigInteger endNs = BigInteger.valueOf(TimeUnit.NANOSECONDS.
+        convert(endMs, TimeUnit.MILLISECONDS));
+    BigInteger deltaNs = endNs.subtract(startNs);
+    if (deltaNs.signum() >= 0) {
+      return TimeUnit.MILLISECONDS.convert(deltaNs.min(
+          BigInteger.valueOf(Long.MAX_VALUE)).longValue(), TimeUnit.NANOSECONDS);
+    }
+    deltaNs = deltaNs.negate();
+    if (deltaNs.compareTo(BigInteger.valueOf(Long.MAX_VALUE / 2)) < 0) {
+      // If the 'startNs' is numerically less than the 'endNs', and the
+      // difference between the two is less than 100 years, it's probably
+      // just clock jitter.  Certain old OSes and CPUs had monotonic clocks
+      // that could go backwards under certain conditions (ironic, given
+      // the name).
+      return 0L;
+    }
+    // Handle rollover.
+    BigInteger revDeltaNs = BigInteger.ONE.shiftLeft(64).subtract(deltaNs);
+    return TimeUnit.MILLISECONDS.convert(revDeltaNs.min(
+        BigInteger.valueOf(Long.MAX_VALUE)).longValue(), TimeUnit.NANOSECONDS);
+  }
+}
diff --git a/htrace-htraced/src/test/java/org/apache/htrace/impl/DataDir.java b/htrace-htraced/src/test/java/org/apache/htrace/impl/DataDir.java
new file mode 100644
index 0000000..f224f6f
--- /dev/null
+++ b/htrace-htraced/src/test/java/org/apache/htrace/impl/DataDir.java
@@ -0,0 +1,58 @@
+/*
+ * 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.htrace.impl;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.UUID;
+
+/**
+ * Small util for making a data directory for tests to use when running tests. We put it up at
+ * target/test-data/UUID.  Create an instance of this class per unit test run and it will take
+ * care of setting up the dirs for you.  Pass what is returned here as location from which to
+ * have daemons and tests dump data.
+ */
+public class DataDir implements Closeable {
+  private final File dir;
+
+  public DataDir() throws IOException {
+    String baseDir = System.getProperty(
+        "test.data.base.dir", "target");
+    File testData = new File(new File(baseDir), "test-data");
+    this.dir = new File(testData, UUID.randomUUID().toString());
+    Files.createDirectories(this.dir.toPath());
+  }
+
+  public File get() {
+    return dir;
+  }
+
+  @Override
+  public void close() throws IOException {
+    /*for (File file : this.dir.listFiles()) {
+      file.delete();
+    }
+    Files.delete(this.dir.toPath()); */
+  }
+
+  @Override
+  public String toString() {
+    return "DataDir{" + dir.getAbsolutePath() + "}";
+  }
+}
\ No newline at end of file
diff --git a/htrace-htraced/src/test/java/org/apache/htrace/util/HTracedProcess.java b/htrace-htraced/src/test/java/org/apache/htrace/impl/HTracedProcess.java
similarity index 60%
rename from htrace-htraced/src/test/java/org/apache/htrace/util/HTracedProcess.java
rename to htrace-htraced/src/test/java/org/apache/htrace/impl/HTracedProcess.java
index 3e800d2..26c1a10 100644
--- a/htrace-htraced/src/test/java/org/apache/htrace/util/HTracedProcess.java
+++ b/htrace-htraced/src/test/java/org/apache/htrace/impl/HTracedProcess.java
@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.htrace.util;
+package org.apache.htrace.impl;
 
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -24,27 +24,76 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
-import java.lang.ProcessBuilder.Redirect;
 import java.net.ServerSocket;
 import java.net.Socket;
-import java.net.URL;
+import java.net.URI;
+import java.nio.file.Paths;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.htrace.core.MilliSpan;
+import org.apache.htrace.core.Span;
+import org.apache.htrace.core.SpanId;
+import org.eclipse.jetty.client.HttpClient;
+import org.eclipse.jetty.client.api.ContentResponse;
+import org.eclipse.jetty.http.HttpStatus;
+import org.junit.Assert;
 
 /**
  * To get instance of HTraced up and running, create an instance of this class.
  * Upon successful construction, htraced is running using <code>dataDir</code> as directory to
  * host data (leveldbs and logs).
- * TODO: We expect to find the htraced in a very particular place. Fragile. Will break if stuff
- * moves.
  */
-public class HTracedProcess extends Process {
+class HTracedProcess extends Process {
   private static final Log LOG = LogFactory.getLog(HTracedProcess.class);
+
+  static class Builder {
+    String host = "localhost";
+
+    Builder() {
+    }
+
+    Builder host(String host) {
+      this.host = host;
+      return this;
+    }
+
+    HTracedProcess build() throws Exception {
+      return new HTracedProcess(this);
+    }
+  }
+
+  /**
+   * Path to the htraced binary.
+   */
+  private final File htracedPath;
+
+  /**
+   * Temporary directory for test files.
+   */
+  private final DataDir dataDir;
+
+  /**
+   * The Java Process object for htraced.
+   */
   private final Process delegate;
 
+  /**
+   * The HTTP host:port returned from htraced.
+   */
   private final String httpAddr;
 
   /**
+   * The HRPC host:port returned from htraced.
+   */
+  private final String hrpcAddr;
+
+  /**
+   * REST client to use to talk to htraced.
+   */
+  private final HttpClient httpClient;
+
+  /**
    * Data send back from the HTraced process on the notification port.
    */
   @JsonIgnoreProperties(ignoreUnknown = true)
@@ -56,42 +105,59 @@
     String httpAddr;
 
     /**
+     * The hostname:port pair which the HTraced process uses for HRPC requests.
+     */
+    @JsonProperty("HrpcAddr")
+    String hrpcAddr;
+
+    /**
      * The process ID of the HTraced process.
      */
     @JsonProperty("ProcessId")
     long processId;
   }
 
-  public HTracedProcess(final File binPath, final File dataDir,
-                        final String host) throws IOException {
+  private HTracedProcess(Builder builder) throws Exception {
+    this.htracedPath = Paths.get(
+        "target", "..", "go", "build", "htraced").toFile();
+    if (!this.htracedPath.exists()) {
+      throw new RuntimeException("No htraced binary exists at " +
+          this.htracedPath);
+    }
+    this.dataDir = new DataDir();
     // Create a notifier socket bound to a random port.
     ServerSocket listener = new ServerSocket(0);
     boolean success = false;
     Process process = null;
+    HttpClient http = null;
     try {
       // Use a random port for the web address.  No 'scheme' yet.
-      String webAddress = host + ":0";
-      String logPath = new File(dataDir, "log.txt").getAbsolutePath();
+      String random = builder.host + ":0";
+      String logPath = new File(dataDir.get(), "log.txt").getAbsolutePath();
       // Pass cmdline args to htraced to it uses our test dir for data.
-      ProcessBuilder pb = new ProcessBuilder(binPath.toString(),
+      ProcessBuilder pb = new ProcessBuilder(htracedPath.getAbsolutePath(),
         "-Dlog.level=TRACE",
         "-Dlog.path=" + logPath,
-        "-Dweb.address=" + webAddress,
+        "-Dweb.address=" + random,
+        "-Dhrpc.address=" + random,
         "-Ddata.store.clear=true",
         "-Dstartup.notification.address=localhost:" + listener.getLocalPort(),
-        "-Ddata.store.directories=" + dataDir.toString());
+        "-Ddata.store.directories=" + dataDir.get().getAbsolutePath());
       pb.redirectErrorStream(true);
       // Inherit STDERR/STDOUT i/o; dumps on console for now.  Can add logs later.
       pb.inheritIO();
-      pb.directory(dataDir);
+      pb.directory(dataDir.get());
       //assert pb.redirectInput() == Redirect.PIPE;
       //assert pb.redirectOutput().file() == dataDir;
       process = pb.start();
       assert process.getInputStream().read() == -1;
       StartupNotificationData data = readStartupNotification(listener);
       httpAddr = data.httpAddr;
+      hrpcAddr = data.hrpcAddr;
       LOG.info("Started htraced process " + data.processId + " with http " +
                "address " + data.httpAddr + ", logging to " + logPath);
+      http = RestBufferManager.createHttpClient(60000L, 60000L);
+      http.start();
       success = true;
     } finally {
       if (!success) {
@@ -100,9 +166,13 @@
           process.destroy();
           process = null;
         }
+        if (http != null) {
+          http.stop();
+        }
       }
       delegate = process;
       listener.close();
+      httpClient = http;
     }
   }
 
@@ -149,7 +219,18 @@
   }
 
   public void destroy() {
+    try {
+      httpClient.stop();
+    } catch (Exception e) {
+      LOG.error("Error stopping httpClient", e);
+    }
     delegate.destroy();
+    try {
+      dataDir.close();
+    } catch (Exception e) {
+      LOG.error("Error closing " + dataDir, e);
+    }
+    LOG.trace("Destroyed htraced process.");
   }
 
   public String toString() {
@@ -160,6 +241,10 @@
     return httpAddr;
   }
 
+  public String getHrpcAddr() {
+    return hrpcAddr;
+  }
+
   /**
    * Ugly but how else to do file-math?
    * @param topLevel Presumes top-level of the htrace checkout.
@@ -169,4 +254,24 @@
     return new File(new File(new File(new File(topLevel, "htrace-htraced"), "go"), "build"),
       "htraced");
   }
+
+  public String getServerInfoJson() throws Exception {
+    ContentResponse response = httpClient.GET(
+        new URI(String.format("http://%s/server/info", httpAddr)));
+    Assert.assertEquals("application/json", response.getMediaType());
+    Assert.assertEquals(HttpStatus.OK_200, response.getStatus());
+    return response.getContentAsString();
+  }
+
+  public Span getSpan(SpanId spanId) throws Exception {
+    ContentResponse response = httpClient.GET(
+        new URI(String.format("http://%s/span/%s",
+            httpAddr, spanId.toString())));
+    Assert.assertEquals("application/json", response.getMediaType());
+    String responseJson = response.getContentAsString().trim();
+    if (responseJson.isEmpty()) {
+      return null;
+    }
+    return MilliSpan.fromJson(responseJson);
+  }
 }
diff --git a/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedRESTReceiver.java b/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedRESTReceiver.java
deleted file mode 100644
index d52f071..0000000
--- a/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedRESTReceiver.java
+++ /dev/null
@@ -1,215 +0,0 @@
-/**
- * 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.htrace.impl;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-
-import java.io.File;
-import java.net.URL;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.htrace.core.HTraceConfiguration;
-import org.apache.htrace.core.MilliSpan;
-import org.apache.htrace.core.Span;
-import org.apache.htrace.core.SpanId;
-import org.apache.htrace.core.TracerId;
-import org.apache.htrace.util.DataDir;
-import org.apache.htrace.util.HTracedProcess;
-import org.apache.htrace.util.TestUtil;
-import org.eclipse.jetty.client.HttpClient;
-import org.eclipse.jetty.client.api.ContentResponse;
-import org.eclipse.jetty.http.HttpStatus;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-
-public class TestHTracedRESTReceiver {
-  private static final Log LOG =
-      LogFactory.getLog(TestHTracedRESTReceiver.class);
-  private URL restServerUrl;
-  private DataDir dataDir;
-  HTracedProcess htraced;
-
-  @Before
-  public void setUp() throws Exception {
-    this.dataDir = new DataDir();
-    File tlDir = DataDir.getTopLevelOfCheckout(this.dataDir.getDataDir());
-    File pathToHTracedBinary = HTracedProcess.getPathToHTraceBinaryFromTopLevel(tlDir);
-    this.htraced = new HTracedProcess(pathToHTracedBinary,
-        dataDir.getDataDir(), "localhost");
-    this.restServerUrl = new URL("http://" + htraced.getHttpAddr() + "/");
-  }
-
-  @After
-  public void tearDown() throws Exception {
-    if (this.htraced != null) this.htraced.destroy();
-  }
-
-  /**
-   * Our simple version of htrace configuration for testing.
-   */
-  private final class TestHTraceConfiguration extends HTraceConfiguration {
-    private final URL restServerUrl;
-    final static String TRACER_ID = "TestHTracedRESTReceiver";
-
-    public TestHTraceConfiguration(final URL restServerUrl) {
-      this.restServerUrl = restServerUrl;
-    }
-
-    @Override
-    public String get(String key) {
-      return null;
-    }
-
-    @Override
-    public String get(String key, String defaultValue) {
-      if (key.equals(HTracedRESTReceiver.HTRACED_REST_URL_KEY)) {
-        return this.restServerUrl.toString();
-      } else if (key.equals(TracerId.TRACER_ID_KEY)) {
-        return TRACER_ID;
-      }
-      return defaultValue;
-    }
-  }
-
-  /**
-   * Make sure the REST server basically works.
-   * @throws Exception
-   */
-  @Test (timeout = 10000)
-  public void testBasicGet() throws Exception {
-    HTracedRESTReceiver receiver =
-      new HTracedRESTReceiver(new TestHTraceConfiguration(this.restServerUrl));
-    HttpClient http = receiver.createHttpClient(60000L, 60000L);
-    http.start();
-    try {
-      // Do basic a GET /server/info against htraced
-      ContentResponse response =
-        http.GET(restServerUrl + "server/info");
-      assertEquals("application/json", response.getMediaType());
-      String content = processGET(response);
-      assertTrue(content.contains("ReleaseVersion"));
-      System.out.println(content);
-    } finally {
-      http.stop();
-      receiver.close();
-    }
-  }
-
-  private String processGET(final ContentResponse response) {
-    assertTrue("" + response.getStatus(), HttpStatus.OK_200 <= response.getStatus() &&
-      response.getStatus() <= HttpStatus.NO_CONTENT_204);
-    return response.getContentAsString();
-  }
-
-  private void testSendingSpansImpl(boolean testClose) throws Exception {
-    final HTracedRESTReceiver receiver =
-      new HTracedRESTReceiver(new TestHTraceConfiguration(this.restServerUrl));
-    final int NUM_SPANS = 3;
-    final HttpClient http = receiver.createHttpClient(60000, 60000);
-    http.start();
-    Span spans[] = new Span[NUM_SPANS];
-    for (int i = 0; i < NUM_SPANS; i++) {
-      MilliSpan.Builder builder = new MilliSpan.Builder().
-          parents(new SpanId[] { new SpanId(1L, 1L) }).
-          spanId(new SpanId(1L, i));
-      if (i == NUM_SPANS - 1) {
-        builder.tracerId("specialTrid");
-      } else {
-        builder.tracerId(TestHTraceConfiguration.TRACER_ID);
-      }
-      spans[i] = builder.build();
-    }
-    try {
-      for (int i = 0; i < NUM_SPANS; i++) {
-        LOG.info("receiving " + spans[i].toString());
-        receiver.receiveSpan(spans[i]);
-      }
-      if (testClose) {
-        receiver.close();
-      } else {
-        receiver.startFlushing();
-      }
-      TestUtil.waitFor(new TestUtil.Supplier<Boolean>() {
-        @Override
-        public Boolean get() {
-          try {
-            for (int i = 0; i < NUM_SPANS; i++) {
-              // This is what the REST server expects when querying for a
-              // span id.
-              String findSpan = String.format("span/%s",
-                  new SpanId(1L, i).toString());
-              ContentResponse response =
-                  http.GET(restServerUrl + findSpan);
-              String content = processGET(response);
-              if ((content == null) || (content.length() == 0)) {
-                LOG.info("Failed to find span " + i);
-                return false;
-              }
-              LOG.info("Got " + content + " for span " + i);
-              MilliSpan dspan = MilliSpan.fromJson(content);
-              assertEquals(new SpanId(1, i).toString(),
-                dspan.getSpanId().toString());
-              // Every span should have the tracer ID we set in the
-              // configuration... except for the last span, which had
-              // a custom value set.
-              if (i == NUM_SPANS - 1) {
-                assertEquals("specialTrid", dspan.getTracerId());
-              } else {
-                assertEquals(TestHTraceConfiguration.TRACER_ID,
-                    dspan.getTracerId());
-              }
-            }
-            return true;
-          } catch (Throwable t) {
-            LOG.error("Got exception", t);
-            return false;
-          }
-        }
-      }, 10, 20000);
-    } finally {
-      http.stop();
-      if (!testClose) {
-        receiver.close();
-      }
-    }
-  }
-
-  /**
-   * Send 100 spans then confirm they made it in.
-   * @throws Exception
-   */
-  @Test (timeout = 60000)
-  public void testSendingSpans() throws Exception {
-    testSendingSpansImpl(false);
-  }
-
-  /**
-   * Test that the REST receiver blocks during shutdown until all spans are sent
-   * (or a long timeout elapses).  Otherwise, short-lived client processes will
-   * never have a chance to send all their spans and we will have incomplete
-   * information.
-   */
-  @Test (timeout = 60000)
-  public void testShutdownBlocksUntilSpanAreSent() throws Exception {
-    testSendingSpansImpl(true);
-  }
-}
diff --git a/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedReceiver.java b/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedReceiver.java
new file mode 100644
index 0000000..99f00a1
--- /dev/null
+++ b/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedReceiver.java
@@ -0,0 +1,572 @@
+/**
+ * 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.htrace.impl;
+
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.Random;
+import java.util.concurrent.Semaphore;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.htrace.core.HTraceConfiguration;
+import org.apache.htrace.core.Span;
+import org.apache.htrace.core.SpanId;
+import org.apache.htrace.core.TracerId;
+import org.apache.htrace.impl.HTracedSpanReceiver.FaultInjector;
+import org.apache.htrace.util.TestUtil;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestRule;
+import org.junit.rules.TestWatcher;
+import org.junit.runner.Description;
+
+public class TestHTracedReceiver {
+  private static final Log LOG = LogFactory.getLog(TestHTracedReceiver.class);
+
+  @BeforeClass
+  public static void beforeClass() {
+    // Allow setting really small buffer sizes for testing purposes.
+    // We do not allow setting such small sizes in production.
+    Conf.BUFFER_SIZE_MIN = 0;
+  }
+
+  @Rule
+  public TestRule watcher = new TestWatcher() {
+    protected void starting(Description description) {
+      LOG.info("*** Starting junit test: " + description.getMethodName());
+    }
+
+    protected void finished(Description description) {
+      LOG.info("*** Finished junit test: " + description.getMethodName());
+    }
+  };
+
+  @Test(timeout = 60000)
+  public void testGetServerInfoJson() throws Exception {
+    HTracedProcess ht = new HTracedProcess.Builder().build();
+    try {
+      String response = ht.getServerInfoJson();
+      assertTrue(response.contains("ReleaseVersion"));
+    } finally {
+      ht.destroy();
+    }
+  }
+
+  private void waitForSpans(final HTracedProcess ht, Span[] spans)
+      throws Exception {
+    waitForSpans(ht, spans, spans.length);
+  }
+
+  private void waitForSpans(final HTracedProcess ht, Span[] spans,
+      int numSpans) throws Exception {
+    final LinkedList<SpanId> spanIds = new LinkedList<SpanId>();
+    for (int i = 0; i < numSpans; i++) {
+      spanIds.add(spans[i].getSpanId());
+    }
+    boolean success = false;
+    try {
+      TestUtil.waitFor(new TestUtil.Supplier<Boolean>() {
+        @Override
+        public Boolean get() {
+          for (Iterator<SpanId> iter = spanIds.iterator();
+               iter.hasNext(); ) {
+            SpanId spanId = iter.next();
+            try {
+              if (ht.getSpan(spanId) == null) {
+                return false;
+              }
+            } catch (InterruptedException e) {
+              LOG.error("Got InterruptedException while looking for " +
+                  "span ID " + spanId, e);
+              Thread.currentThread().interrupt();
+            } catch (Exception e) {
+              LOG.error("Got error looking for span ID " + spanId, e);
+              return false;
+            }
+            iter.remove();
+          }
+          return true;
+        }
+      }, 10, 30000);
+      success = true;
+    } finally {
+      if (!success) {
+        String prefix = "";
+        StringBuilder idStringBld = new StringBuilder();
+        for (Iterator<SpanId> iter = spanIds.iterator();
+             iter.hasNext(); ) {
+          idStringBld.append(prefix);
+          idStringBld.append(iter.next());
+          prefix = ",";
+        }
+        LOG.error("Unable to find span IDs " + idStringBld.toString());
+      }
+    }
+  }
+
+  /**
+   * Test that we can send spans via the HRPC interface.
+   */
+  @Test(timeout = 10000) //60000)
+  public void testSendSpansViaPacked() throws Exception {
+    final Random rand = new Random(123);
+    final HTracedProcess ht = new HTracedProcess.Builder().build();
+    try {
+      HTraceConfiguration conf = HTraceConfiguration.fromMap(
+          new HashMap<String, String>() {{
+            put(TracerId.TRACER_ID_KEY, "testSendSpansViaPacked");
+            put(Conf.ADDRESS_KEY, ht.getHrpcAddr());
+            put(Conf.PACKED_KEY, "true");
+            put(Conf.MAX_FLUSH_INTERVAL_MS_KEY, "100");
+            put(Conf.ERROR_LOG_PERIOD_MS_KEY, "0");
+          }});
+      HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf);
+      Span[] spans = TestUtil.randomSpans(rand, 10);
+      for (Span span : spans) {
+        rcvr.receiveSpan(span);
+      }
+      waitForSpans(ht, spans);
+      rcvr.close();
+    } finally {
+      ht.destroy();
+    }
+  }
+
+  /**
+   * Test that when the SpanReceiver is closed, we send any spans we have
+   * buffered via the HRPC interface.
+   */
+  @Test(timeout = 60000)
+  public void testSendSpansViaPackedAndClose() throws Exception {
+    final Random rand = new Random(456);
+    final HTracedProcess ht = new HTracedProcess.Builder().build();
+    try {
+      HTraceConfiguration conf = HTraceConfiguration.fromMap(
+          new HashMap<String, String>() {{
+            put(TracerId.TRACER_ID_KEY, "testSendSpansViaPackedAndClose");
+            put(Conf.ADDRESS_KEY, ht.getHrpcAddr());
+            put(Conf.PACKED_KEY, "true");
+            put(Conf.MAX_FLUSH_INTERVAL_MS_KEY, "60000");
+          }});
+      HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf);
+      Span[] spans = TestUtil.randomSpans(rand, 10);
+      for (Span span : spans) {
+        rcvr.receiveSpan(span);
+      }
+      rcvr.close();
+      waitForSpans(ht, spans);
+    } finally {
+      ht.destroy();
+    }
+  }
+
+  /**
+   * Test that we can send spans via the REST interface.
+   */
+  @Test(timeout = 60000)
+  public void testSendSpansViaRest() throws Exception {
+    final Random rand = new Random(789);
+    final HTracedProcess ht = new HTracedProcess.Builder().build();
+    try {
+      HTraceConfiguration conf = HTraceConfiguration.fromMap(
+          new HashMap<String, String>() {{
+            put(TracerId.TRACER_ID_KEY, "testSendSpansViaRest");
+            put(Conf.ADDRESS_KEY, ht.getHttpAddr());
+            put(Conf.PACKED_KEY, "false");
+            put(Conf.MAX_FLUSH_INTERVAL_MS_KEY, "100");
+          }});
+      HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf);
+      Span[] spans = TestUtil.randomSpans(rand, 10);
+      for (Span span : spans) {
+        rcvr.receiveSpan(span);
+      }
+      waitForSpans(ht, spans);
+      rcvr.close();
+    } finally {
+      ht.destroy();
+    }
+  }
+
+  /**
+   * Test that when the SpanReceiver is closed, we send any spans we have
+   * buffered via the REST interface.
+   */
+  @Test(timeout = 60000)
+  public void testSendSpansViaRestAndClose() throws Exception {
+    final Random rand = new Random(321);
+    final HTracedProcess ht = new HTracedProcess.Builder().build();
+    try {
+      HTraceConfiguration conf = HTraceConfiguration.fromMap(
+          new HashMap<String, String>() {{
+            put(TracerId.TRACER_ID_KEY, "testSendSpansViaRestAndClose");
+            put(Conf.ADDRESS_KEY, ht.getHttpAddr());
+            put(Conf.PACKED_KEY, "false");
+            put(Conf.MAX_FLUSH_INTERVAL_MS_KEY, "60000");
+          }});
+      HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf);
+      Span[] spans = TestUtil.randomSpans(rand, 10);
+      for (Span span : spans) {
+        rcvr.receiveSpan(span);
+      }
+      rcvr.close();
+      waitForSpans(ht, spans);
+    } finally {
+      ht.destroy();
+    }
+  }
+
+  private static class Mutable<T> {
+    private T t;
+
+    Mutable(T t) {
+      this.t = t;
+    }
+
+    void set(T t) {
+      this.t = t;
+    }
+
+    T get() {
+      return this.t;
+    }
+  }
+
+  private static class TestHandleContentLengthTriggerInjector
+      extends HTracedSpanReceiver.FaultInjector {
+    final Semaphore threadStartSem = new Semaphore(0);
+    int contentLengthOnTrigger = 0;
+
+    @Override
+    public synchronized void handleContentLengthTrigger(int len) {
+      contentLengthOnTrigger = len;
+    }
+    @Override
+    public void handleThreadStart() throws Exception {
+      threadStartSem.acquire();
+    }
+
+    public synchronized int getContentLengthOnTrigger() {
+      return contentLengthOnTrigger;
+    }
+  }
+
+  /**
+   * Test that filling up one of the buffers causes us to trigger a flush and
+   * start using the other buffer, when using PackedBufferManager.
+   * This also tests that PackedBufferManager can correctly handle a buffer
+   * getting full.
+   */
+  @Test(timeout = 60000)
+  public void testFullBufferCausesPackedThreadTrigger() throws Exception {
+    final Random rand = new Random(321);
+    final HTracedProcess ht = new HTracedProcess.Builder().build();
+    try {
+      HTraceConfiguration conf = HTraceConfiguration.fromMap(
+          new HashMap<String, String>() {{
+            put(TracerId.TRACER_ID_KEY,
+                "testFullBufferCausesPackedThreadTrigger");
+            put(Conf.ADDRESS_KEY, ht.getHrpcAddr());
+            put(Conf.PACKED_KEY, "true");
+            put(Conf.BUFFER_SIZE_KEY, "16384");
+            put(Conf.BUFFER_SEND_TRIGGER_FRACTION_KEY, "0.95");
+          }});
+      TestHandleContentLengthTriggerInjector injector =
+          new TestHandleContentLengthTriggerInjector();
+      HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf, injector);
+      Span[] spans = TestUtil.randomSpans(rand, 47);
+      for (Span span : spans) {
+        rcvr.receiveSpan(span);
+      }
+      Assert.assertTrue("The wakePostSpansThread should have been " +
+          "triggered by the spans added so far.  " +
+          "contentLengthOnTrigger = " + injector.getContentLengthOnTrigger(),
+          injector.getContentLengthOnTrigger() > 16000);
+      injector.threadStartSem.release();
+      rcvr.close();
+      waitForSpans(ht, spans, 45);
+    } finally {
+      ht.destroy();
+    }
+  }
+
+  /**
+   * Test that filling up one of the buffers causes us to trigger a flush and
+   * start using the other buffer, when using RestBufferManager.
+   * This also tests that RestBufferManager can correctly handle a buffer
+   * getting full.
+   */
+  @Test(timeout = 60000)
+  public void testFullBufferCausesRestThreadTrigger() throws Exception {
+    final Random rand = new Random(321);
+    final HTracedProcess ht = new HTracedProcess.Builder().build();
+    try {
+      HTraceConfiguration conf = HTraceConfiguration.fromMap(
+          new HashMap<String, String>() {{
+            put(TracerId.TRACER_ID_KEY,
+                "testFullBufferCausesRestThreadTrigger");
+            put(Conf.ADDRESS_KEY, ht.getHttpAddr());
+            put(Conf.PACKED_KEY, "false");
+            put(Conf.BUFFER_SIZE_KEY, "16384");
+            put(Conf.BUFFER_SEND_TRIGGER_FRACTION_KEY, "0.95");
+          }});
+      TestHandleContentLengthTriggerInjector injector =
+          new TestHandleContentLengthTriggerInjector();
+      HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf, injector);
+      Span[] spans = TestUtil.randomSpans(rand, 34);
+      for (Span span : spans) {
+        rcvr.receiveSpan(span);
+      }
+      Assert.assertTrue("The wakePostSpansThread should have been " +
+              "triggered by the spans added so far.  " +
+              "contentLengthOnTrigger = " + injector.getContentLengthOnTrigger(),
+          injector.getContentLengthOnTrigger() > 16000);
+      injector.threadStartSem.release();
+      rcvr.close();
+      waitForSpans(ht, spans, 33);
+    } finally {
+      ht.destroy();
+    }
+  }
+
+  /**
+   * A FaultInjector that causes all flushes to fail until a specified
+   * number of milliseconds have passed.
+   */
+  private static class TestInjectFlushFaults
+      extends HTracedSpanReceiver.FaultInjector {
+    private long remainingFaults;
+
+    TestInjectFlushFaults(long remainingFaults) {
+      this.remainingFaults = remainingFaults;
+    }
+
+    @Override
+    public synchronized void handleFlush() throws IOException {
+      if (remainingFaults > 0) {
+        remainingFaults--;
+        throw new IOException("Injected IOException into flush " +
+            "code path.");
+      }
+    }
+  }
+
+  /**
+   * Test that even if the flush fails, the system stays stable and we can
+   * still close the span receiver.
+   */
+  @Test(timeout = 60000)
+  public void testPackedThreadHandlesFlushFailure() throws Exception {
+    final Random rand = new Random(321);
+    final HTracedProcess ht = new HTracedProcess.Builder().build();
+    try {
+      HTraceConfiguration conf = HTraceConfiguration.fromMap(
+          new HashMap<String, String>() {{
+            put(TracerId.TRACER_ID_KEY, "testPackedThreadHandlesFlushFailure");
+            put(Conf.ADDRESS_KEY, ht.getHrpcAddr());
+            put(Conf.PACKED_KEY, "true");
+          }});
+      TestInjectFlushFaults injector = new TestInjectFlushFaults(Long.MAX_VALUE);
+      HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf, injector);
+      Span[] spans = TestUtil.randomSpans(rand, 15);
+      for (Span span : spans) {
+        rcvr.receiveSpan(span);
+      }
+      rcvr.close();
+    } finally {
+      ht.destroy();
+    }
+  }
+
+  /**
+   * Test that even if the flush fails, the system stays stable and we can
+   * still close the span receiver.
+   */
+  @Test(timeout = 60000)
+  public void testRestThreadHandlesFlushFailure() throws Exception {
+    final Random rand = new Random(321);
+    final HTracedProcess ht = new HTracedProcess.Builder().build();
+    try {
+      HTraceConfiguration conf = HTraceConfiguration.fromMap(
+          new HashMap<String, String>() {{
+            put(TracerId.TRACER_ID_KEY, "testRestThreadHandlesFlushFailure");
+            put(Conf.ADDRESS_KEY, ht.getHttpAddr());
+            put(Conf.PACKED_KEY, "false");
+          }});
+      TestInjectFlushFaults injector = new TestInjectFlushFaults(Long.MAX_VALUE);
+      HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf, injector);
+      Span[] spans = TestUtil.randomSpans(rand, 15);
+      for (Span span : spans) {
+        rcvr.receiveSpan(span);
+      }
+      rcvr.close();
+    } finally {
+      ht.destroy();
+    }
+  }
+
+  /**
+   * A FaultInjector that causes all flushes to fail until a specified
+   * number of milliseconds have passed.
+   */
+  private static class WaitForFlushes
+      extends HTracedSpanReceiver.FaultInjector {
+    final Semaphore flushSem;
+
+    WaitForFlushes(int numFlushes) {
+      this.flushSem = new Semaphore(-numFlushes);
+    }
+
+    @Override
+    public void handleFlush() throws IOException {
+      flushSem.release();
+    }
+  }
+
+  /**
+   * Test that the packed code works when performing multiple flushes.
+   */
+  @Test(timeout = 60000)
+  public void testMultiplePackedFlushes() throws Exception {
+    final Random rand = new Random(123);
+    final HTracedProcess ht = new HTracedProcess.Builder().build();
+    try {
+      HTraceConfiguration conf = HTraceConfiguration.fromMap(
+          new HashMap<String, String>() {{
+            put(TracerId.TRACER_ID_KEY, "testMultiplePackedFlushes");
+            put(Conf.ADDRESS_KEY, ht.getHrpcAddr());
+            put(Conf.PACKED_KEY, "true");
+            put(Conf.MAX_FLUSH_INTERVAL_MS_KEY, "1");
+          }});
+      WaitForFlushes injector = new WaitForFlushes(5);
+      HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf, injector);
+      Span[] spans = TestUtil.randomSpans(rand, 3);
+      while (true) {
+        for (Span span : spans) {
+          rcvr.receiveSpan(span);
+        }
+        if (injector.flushSem.availablePermits() >= 0) {
+          break;
+        }
+        Thread.sleep(1);
+      }
+      waitForSpans(ht, spans, 3);
+      rcvr.close();
+    } finally {
+      ht.destroy();
+    }
+  }
+
+  /**
+   * Test that the REST code works when performing multiple flushes.
+   */
+  @Test(timeout = 60000)
+  public void testMultipleRestFlushes() throws Exception {
+    final Random rand = new Random(123);
+    final HTracedProcess ht = new HTracedProcess.Builder().build();
+    try {
+      HTraceConfiguration conf = HTraceConfiguration.fromMap(
+          new HashMap<String, String>() {{
+            put(TracerId.TRACER_ID_KEY, "testMultipleRestFlushes");
+            put(Conf.ADDRESS_KEY, ht.getHttpAddr());
+            put(Conf.PACKED_KEY, "false");
+            put(Conf.MAX_FLUSH_INTERVAL_MS_KEY, "1");
+          }});
+      WaitForFlushes injector = new WaitForFlushes(5);
+      HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf, injector);
+      Span[] spans = TestUtil.randomSpans(rand, 3);
+      while (true) {
+        for (Span span : spans) {
+          rcvr.receiveSpan(span);
+        }
+        if (injector.flushSem.availablePermits() >= 0) {
+          break;
+        }
+        Thread.sleep(1);
+      }
+      waitForSpans(ht, spans, 3);
+      rcvr.close();
+    } finally {
+      ht.destroy();
+    }
+  }
+
+  /**
+   * Test that the packed code works when performing multiple flushes.
+   */
+  @Test(timeout = 60000)
+  public void testPackedRetryAfterFlushError() throws Exception {
+    final Random rand = new Random(123);
+    final HTracedProcess ht = new HTracedProcess.Builder().build();
+    try {
+      HTraceConfiguration conf = HTraceConfiguration.fromMap(
+          new HashMap<String, String>() {{
+            put(TracerId.TRACER_ID_KEY, "testPackedRetryAfterFlushError");
+            put(Conf.ADDRESS_KEY, ht.getHrpcAddr());
+            put(Conf.PACKED_KEY, "true");
+            put(Conf.MAX_FLUSH_INTERVAL_MS_KEY, "1000");
+            put(Conf.FLUSH_RETRY_DELAYS_KEY, "100,100,100,100,100,100,100");
+          }});
+      TestInjectFlushFaults injector = new TestInjectFlushFaults(5);
+      HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf, injector);
+      Span[] spans = TestUtil.randomSpans(rand, 3);
+      for (Span span : spans) {
+        rcvr.receiveSpan(span);
+      }
+      waitForSpans(ht, spans);
+      rcvr.close();
+    } finally {
+      ht.destroy();
+    }
+  }
+
+  /**
+   * Test that the REST code works when performing multiple flushes.
+   */
+  @Test(timeout = 60000)
+  public void testRestRetryAfterFlushError() throws Exception {
+    final Random rand = new Random(123);
+    final HTracedProcess ht = new HTracedProcess.Builder().build();
+    try {
+      HTraceConfiguration conf = HTraceConfiguration.fromMap(
+          new HashMap<String, String>() {{
+            put(TracerId.TRACER_ID_KEY, "testRestRetryAfterFlushError");
+            put(Conf.ADDRESS_KEY, ht.getHttpAddr());
+            put(Conf.PACKED_KEY, "false");
+            put(Conf.MAX_FLUSH_INTERVAL_MS_KEY, "1000");
+            put(Conf.FLUSH_RETRY_DELAYS_KEY, "100,100,100,100,100,100,100");
+          }});
+      TestInjectFlushFaults injector = new TestInjectFlushFaults(5);
+      HTracedSpanReceiver rcvr = new HTracedSpanReceiver(conf, injector);
+      Span[] spans = TestUtil.randomSpans(rand, 3);
+      for (Span span : spans) {
+        rcvr.receiveSpan(span);
+      }
+      waitForSpans(ht, spans);
+      rcvr.close();
+    } finally {
+      ht.destroy();
+    }
+  }
+}
diff --git a/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedReceiverConf.java b/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedReceiverConf.java
new file mode 100644
index 0000000..bf038f1
--- /dev/null
+++ b/htrace-htraced/src/test/java/org/apache/htrace/impl/TestHTracedReceiverConf.java
@@ -0,0 +1,91 @@
+/**
+ * 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.htrace.impl;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.htrace.core.HTraceConfiguration;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestHTracedReceiverConf {
+  private static final Log LOG =
+      LogFactory.getLog(TestHTracedReceiverConf.class);
+
+  @Test(timeout = 60000)
+  public void testParseHostPort() throws Exception {
+    InetSocketAddress addr = new Conf(
+        HTraceConfiguration.fromKeyValuePairs(
+          Conf.ADDRESS_KEY, "example.com:8080")).endpoint;
+    Assert.assertEquals("example.com", addr.getHostName());
+    Assert.assertEquals(8080, addr.getPort());
+
+    addr = new Conf(
+        HTraceConfiguration.fromKeyValuePairs(
+          Conf.ADDRESS_KEY, "127.0.0.1:8081")).endpoint;
+    Assert.assertEquals("127.0.0.1", addr.getHostName());
+    Assert.assertEquals(8081, addr.getPort());
+
+    addr = new Conf(
+        HTraceConfiguration.fromKeyValuePairs(
+          Conf.ADDRESS_KEY, "[ff02:0:0:0:0:0:0:12]:9095")).endpoint;
+    Assert.assertEquals("ff02:0:0:0:0:0:0:12", addr.getHostName());
+    Assert.assertEquals(9095, addr.getPort());
+  }
+
+  private static void verifyFail(String hostPort) {
+    try {
+      new Conf(HTraceConfiguration.fromKeyValuePairs(
+            Conf.ADDRESS_KEY, hostPort));
+      Assert.fail("Expected bad host:port configuration " + hostPort +
+          " to fail, but it succeeded.");
+    } catch (IOException e) {
+      // expected
+    }
+  }
+
+  @Test(timeout = 60000)
+  public void testFailToParseHostPort() throws Exception {
+    verifyFail("localhost"); // no port
+    verifyFail("127.0.0.1"); // no port
+    verifyFail(":8080"); // no hostname
+    verifyFail("bob[ff02:0:0:0:0:0:0:12]:9095"); // bracket at incorrect place
+  }
+
+  @Test(timeout = 60000)
+  public void testGetIntArray() throws Exception {
+    int[] arr = Conf.getIntArray("");
+    Assert.assertEquals(0, arr.length);
+    arr = Conf.getIntArray("123");
+    Assert.assertEquals(1, arr.length);
+    Assert.assertEquals(123, arr[0]);
+    arr = Conf.getIntArray("1,2,3");
+    Assert.assertEquals(3, arr.length);
+    Assert.assertEquals(1, arr[0]);
+    Assert.assertEquals(2, arr[1]);
+    Assert.assertEquals(3, arr[2]);
+    arr = Conf.getIntArray(",-4,5,66,");
+    Assert.assertEquals(3, arr.length);
+    Assert.assertEquals(-4, arr[0]);
+    Assert.assertEquals(5, arr[1]);
+    Assert.assertEquals(66, arr[2]);
+  }
+}
diff --git a/htrace-htraced/src/test/java/org/apache/htrace/impl/TestPackedBuffer.java b/htrace-htraced/src/test/java/org/apache/htrace/impl/TestPackedBuffer.java
new file mode 100644
index 0000000..ed7d904
--- /dev/null
+++ b/htrace-htraced/src/test/java/org/apache/htrace/impl/TestPackedBuffer.java
@@ -0,0 +1,76 @@
+/**
+ * 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.htrace.impl;
+
+import java.nio.ByteBuffer;
+import java.util.Random;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.htrace.core.Span;
+import org.apache.htrace.util.TestUtil;
+import org.junit.Assert;
+import org.junit.Test;
+import org.msgpack.core.MessagePack;
+import org.msgpack.core.MessageUnpacker;
+
+public class TestPackedBuffer {
+  private static final Log LOG = LogFactory.getLog(TestPackedBuffer.class);
+
+  @Test(timeout = 60000)
+  public void testWriteReqFrame() throws Exception {
+    byte[] arr = new byte[PackedBuffer.HRPC_REQ_FRAME_LENGTH];
+    ByteBuffer bb = ByteBuffer.wrap(arr);
+    PackedBuffer buf = new PackedBuffer(bb);
+    PackedBuffer.writeReqFrame(bb, 1, 123, 456);
+    Assert.assertEquals(PackedBuffer.HRPC_REQ_FRAME_LENGTH, bb.position());
+    Assert.assertEquals("48 54 52 43 " +
+        "01 00 00 00 " +
+        "7b 00 00 00 00 00 00 00 " +
+        "c8 01 00 00",
+        buf.toHexString());
+  }
+
+  @Test(timeout = 60000)
+  public void testPackSpans() throws Exception {
+    Random rand = new Random(123);
+    byte[] arr = new byte[16384];
+    ByteBuffer bb = ByteBuffer.wrap(arr);
+    bb.limit(bb.capacity());
+    PackedBuffer buf = new PackedBuffer(bb);
+    final int NUM_TEST_SPANS = 5;
+    Span[] spans = new Span[NUM_TEST_SPANS];
+    for (int i = 0; i < NUM_TEST_SPANS; i++) {
+      spans[i] = TestUtil.randomSpan(rand);
+    }
+    for (int i = 0; i < NUM_TEST_SPANS; i++) {
+      buf.writeSpan(spans[i]);
+    }
+    LOG.info("wrote " + buf.toHexString());
+    MessagePack msgpack = new MessagePack(PackedBuffer.MSGPACK_CONF);
+    MessageUnpacker unpacker = msgpack.newUnpacker(arr, 0, bb.position());
+    Span[] respans = new Span[NUM_TEST_SPANS];
+    for (int i = 0; i < NUM_TEST_SPANS; i++) {
+      respans[i] = PackedBuffer.readSpan(unpacker);
+    }
+    for (int i = 0; i < NUM_TEST_SPANS; i++) {
+      Assert.assertEquals("Failed to read back span " + i,
+          spans[i].toJson(), respans[i].toJson());
+    }
+  }
+}
diff --git a/htrace-htraced/src/test/java/org/apache/htrace/impl/TestTimeUtil.java b/htrace-htraced/src/test/java/org/apache/htrace/impl/TestTimeUtil.java
new file mode 100644
index 0000000..630a02a
--- /dev/null
+++ b/htrace-htraced/src/test/java/org/apache/htrace/impl/TestTimeUtil.java
@@ -0,0 +1,46 @@
+/**
+ * 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.htrace.impl;
+
+import static org.junit.Assert.assertTrue;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestTimeUtil {
+  /**
+   * Test that our deltaMs function can compute the time difference between any
+   * two monotonic times in milliseconds.
+   */
+  @Test(timeout = 60000)
+  public void testDeltaMs() throws Exception {
+    Assert.assertEquals(0, TimeUtil.deltaMs(0, 0));
+    Assert.assertEquals(1, TimeUtil.deltaMs(0, 1));
+    Assert.assertEquals(0, TimeUtil.deltaMs(1, 0));
+    Assert.assertEquals(10, TimeUtil.deltaMs(1000, 1010));
+    long minMs = TimeUnit.MILLISECONDS.convert(Long.MIN_VALUE,
+        TimeUnit.NANOSECONDS);
+    long maxMs = TimeUnit.MILLISECONDS.convert(Long.MAX_VALUE,
+        TimeUnit.NANOSECONDS);
+    Assert.assertEquals(10, TimeUtil.deltaMs(minMs, minMs + 10));
+    Assert.assertEquals(maxMs, TimeUtil.deltaMs(minMs, maxMs));
+    Assert.assertEquals(11, TimeUtil.deltaMs(maxMs - 10, minMs));
+  }
+}
diff --git a/htrace-htraced/src/test/java/org/apache/htrace/util/DataDir.java b/htrace-htraced/src/test/java/org/apache/htrace/util/DataDir.java
deleted file mode 100644
index 74731fa..0000000
--- a/htrace-htraced/src/test/java/org/apache/htrace/util/DataDir.java
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
- * 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.htrace.util;
-
-import java.io.File;
-import java.io.IOException;
-import java.util.UUID;
-
-/**
- * Small util for making a data directory for tests to use when running tests. We put it up at
- * target/test-data/UUID.  Create an instance of this class per unit test run and it will take
- * care of setting up the dirs for you.  Pass what is returned here as location from which to
- * have daemons and tests dump data.
- * TODO: Add close on exit.
- */
-public class DataDir {
-  private File baseTestDir = null;
-  private File testDir = null;
-
-  /**
-   * System property key to get base test directory value
-   */
-  public static final String TEST_BASE_DIRECTORY_KEY = "test.data.base.dir";
-
-  /**
-   * Default base directory for test output.
-   */
-  public static final String TEST_BASE_DIRECTORY_DEFAULT = "target";
-
-  public static final String TEST_BASE_DIRECTORY_NAME = "test-data";
-
-  /**
-   * @return Where to write test data on local filesystem; usually
-   * {@link #TEST_BASE_DIRECTORY_DEFAULT}
-   * Should not be used directly by the unit tests, hence its's private.
-   * Unit test will use a subdirectory of this directory.
-   * @see #setupDataTestDir()
-   */
-  private synchronized File getBaseTestDir() {
-    if (this.baseTestDir != null) return this.baseTestDir;
-    String testHome = System.getProperty(TEST_BASE_DIRECTORY_KEY, TEST_BASE_DIRECTORY_DEFAULT);
-    this.baseTestDir = new File(testHome, TEST_BASE_DIRECTORY_NAME);
-    return this.baseTestDir;
-  }
-
-  /**
-   * @return Absolute path to the dir created by this instance.
-   * @throws IOException 
-   */
-  public synchronized File getDataDir() throws IOException {
-    if (this.testDir != null) return this.testDir;
-    this.testDir = new File(getBaseTestDir(), UUID.randomUUID().toString());
-    if (!this.testDir.exists()) {
-      if (!this.testDir.mkdirs()) throw new IOException("Failed mkdirs for " + this.testDir);
-    }
-    // Return absolute path. A relative passed to htraced will have it create data dirs relative
-    // to its data dir rather than in it.
-    return this.testDir.getAbsoluteFile();
-  }
-
-  /**
-   * Fragile. Ugly. Presumes paths. Best we can do for now until htraced comes local to this module
-   * and is moved out of src dir.
-   * @param dataDir A datadir gotten from {@link #getDataDir()}
-   * @return Top-level of the checkout.
-   */
-  public static File getTopLevelOfCheckout(final File dataDir) {
-    // Need absolute else we run out of road when dir is relative to this module.
-    File absolute = dataDir.getAbsoluteFile();
-    // Check we are where we think we are.
-    File testDataDir = absolute.getParentFile();
-    if (!testDataDir.getName().equals(TEST_BASE_DIRECTORY_NAME)) {
-      throw new IllegalArgumentException(dataDir.toString());
-    }
-    // Do another check.
-    File targetDir = testDataDir.getParentFile();
-    if (!targetDir.getName().equals(TEST_BASE_DIRECTORY_DEFAULT)) {
-      throw new IllegalArgumentException(dataDir.toString());
-    }
-    // Back up last two dirs out of the htrace-htraced dir.
-    return targetDir.getParentFile().getParentFile();
-  }
-}
\ No newline at end of file
diff --git a/htrace-htraced/src/test/java/org/apache/htrace/util/TestHTracedProcess.java b/htrace-htraced/src/test/java/org/apache/htrace/util/TestHTracedProcess.java
deleted file mode 100644
index 67e3a21..0000000
--- a/htrace-htraced/src/test/java/org/apache/htrace/util/TestHTracedProcess.java
+++ /dev/null
@@ -1,100 +0,0 @@
-/*
- * 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.htrace.util;
-
-import static org.junit.Assert.assertTrue;
-
-import java.io.BufferedReader;
-import java.io.File;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.net.URL;
-import java.net.URLConnection;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.junit.Before;
-import org.junit.Test;
-
-/**
- * Test putting up an htraced and making sure it basically works.
- * Makes presumption about paths; where data is relative to the htraced binary, etc., encoded
- * in methods in the below.
- */
-public class TestHTracedProcess {
-  private static final Log LOG =
-      LogFactory.getLog(TestHTracedProcess.class);
-  private DataDir testDir = null;
-  private final int TIMEOUT = 10000;
-
-  @Before
-  public void setupTest() {
-    this.testDir = new DataDir();
-  }
-
-  /*
-   * Do a basic GET of the server info from the running htraced instance.
-   */
-  private String doGet(final URL url) throws IOException {
-    URLConnection connection = url.openConnection();
-    connection.setConnectTimeout(TIMEOUT);
-    connection.setReadTimeout(TIMEOUT);
-    connection.connect();
-    StringBuffer sb = new StringBuffer();
-    BufferedReader reader = new BufferedReader(
-        new InputStreamReader(connection.getInputStream()));
-    try {
-      String line = null;
-      while ((line = reader.readLine()) != null) {
-        System.out.println(line);
-        sb.append(line);
-      }
-    } finally {
-      reader.close();
-    }
-    return sb.toString();
-  }
-
-  /**
-   * Put up an htraced instance and do a Get against /server/info.
-   * @throws IOException
-   * @throws InterruptedException
-   */
-  @Test (timeout=10000)
-  public void testStartStopHTraced() throws Exception {
-    HTracedProcess htraced = null;
-    File dataDir = this.testDir.getDataDir();
-    File topLevel = DataDir.getTopLevelOfCheckout(dataDir);
-    try {
-      htraced = new HTracedProcess(HTracedProcess.
-          getPathToHTraceBinaryFromTopLevel(topLevel),
-          dataDir, "localhost");
-      LOG.info("Started HTracedProcess with REST server URL " +
-          htraced.getHttpAddr());
-      String str = doGet(new URL(
-          "http://" + htraced.getHttpAddr() + "/server/info"));
-      // Assert we go something back.
-      assertTrue(str.contains("ReleaseVersion"));
-      // Assert that the datadir is not empty.
-    } finally {
-      if (htraced != null) {
-        htraced.destroy();
-        System.out.println("ExitValue=" + htraced.waitFor());
-      }
-    }
-  }
-}
\ No newline at end of file