diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplate.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplate.java
index 3edcbedd9..64d061984 100644
--- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplate.java
+++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplate.java
@@ -31,6 +31,7 @@ import org.springframework.web.client.ResourceAccessException;
 import org.springframework.web.client.RestTemplate;
 
 import java.io.IOException;
+import java.net.URI;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -112,6 +113,27 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck
         callback));
   }
 
+  public <B, V> Future<V> put(B bucket, V value, AsyncKeyValueStoreOperation<V> callback) {
+    return put(bucket, value, null, null, callback);
+  }
+
+  public <B, V> Future<V> put(B bucket, V value, Map<String, String> metaData, AsyncKeyValueStoreOperation<V> callback) {
+    return put(bucket, value, metaData, null, callback);
+  }
+
+  @SuppressWarnings({"unchecked"})
+  public <B, V> Future<V> put(B bucket, V value, Map<String, String> metaData, QosParameters qosParams, AsyncKeyValueStoreOperation<V> callback) {
+    Assert.notNull(bucket, "Bucket cannot be null");
+    String bucketName = (null != qosParams ? bucket.toString() + extractQosParameters(qosParams) : bucket
+        .toString());
+
+    HttpHeaders headers = defaultHeaders(metaData);
+    headers.setContentType(extractMediaType(value));
+    headers.set(RIAK_META_CLASSNAME, value.getClass().getName());
+    HttpEntity<V> entity = new HttpEntity<V>(value, headers);
+    return (Future<V>) workerPool.submit(new AsyncPut<V>(bucketName, entity, callback));
+  }
+
   public <B, K, V> Future<?> get(B bucket, K key, AsyncKeyValueStoreOperation<V> callback) {
     return getWithMetaData(bucket, key, null, callback);
   }
@@ -121,7 +143,10 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck
     HttpHeaders headers;
     try {
       headers = restTemplate.headForHeaders(defaultUri, bucket, key);
-      return extractMetaData(headers);
+      RiakMetaData meta = extractMetaData(headers);
+      meta.setBucket((null != bucket ? bucket.toString() : null));
+      meta.setKey((null != key ? key.toString() : null));
+      return meta;
     } catch (ResourceAccessException e) {
     } catch (IOException e) {
       throw new DataAccessResourceFailureException(e.getMessage(), e);
@@ -129,11 +154,36 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck
     return null;
   }
 
+  @SuppressWarnings({"unchecked"})
+  public <B> Future<?> getBucketSchema(B bucket, QosParameters qosParams, final AsyncKeyValueStoreOperation<Map<String, Object>> callback) {
+    Assert.notNull(bucket, "Bucket cannot be null");
+    Assert.notNull(callback, "Callback cannot be null");
+
+    String bucketName = (null != qosParams ? bucket.toString() + extractQosParameters(qosParams) : bucket
+        .toString());
+
+    return workerPool.submit(new AsyncGet(bucketName,
+        "?keys=true",
+        Map.class,
+        new AsyncKeyValueStoreOperation<Object>() {
+          @SuppressWarnings({"unchecked"})
+          public void completed(KeyValueStoreMetaData meta, Object result) {
+            callback.completed(meta, (Map<String, Object>) result);
+          }
+
+          public void failed(Throwable error) {
+            callback.failed(error);
+          }
+        }));
+  }
+
   @SuppressWarnings({"unchecked"})
   public <B, K, T> Future<?> getWithMetaData(B bucket, K key, Class<T> requiredType, AsyncKeyValueStoreOperation<T> callback) {
     String bucketName = (null != bucket ? bucket.toString() : requiredType.getName());
     // Get a key name that may or may not include the QOS parameters.
-    Assert.notNull(key, "Cannot use a <NULL> key.");
+    Assert.notNull(key, "Cannot use a null key.");
+    Assert.notNull(callback, "Callback cannot be null");
+
     if (null == requiredType) {
       requiredType = (Class<T>) getType(bucketName, key.toString());
     }
@@ -251,6 +301,43 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck
     return setWithMetaData(bucket, key, value, metaData, null, callback);
   }
 
+  protected class AsyncPut<V> implements Runnable {
+
+    private String bucket;
+    private HttpEntity<V> entity = null;
+    private AsyncKeyValueStoreOperation<V> callback = null;
+
+    public AsyncPut(String bucket, HttpEntity<V> entity, AsyncKeyValueStoreOperation<V> callback) {
+      this.bucket = bucket;
+      this.entity = entity;
+      this.callback = callback;
+    }
+
+    public void run() {
+      try {
+        URI location = getRestTemplate().postForLocation(defaultUri, entity, bucket, "");
+        String path = location.getPath();
+        String key = path.substring(path.lastIndexOf("/") + 1);
+
+        HttpHeaders headers = getRestTemplate().headForHeaders(defaultUri, bucket, key);
+        if (null != callback) {
+          RiakMetaData meta = extractMetaData(headers);
+          meta.setBucket((null != bucket ? bucket.toString() : null));
+          meta.setKey((null != key ? key.toString() : null));
+          callback.completed(meta, entity.getBody());
+        }
+      } catch (Throwable t) {
+        DataStoreOperationException dsoe = new DataStoreOperationException(t.getMessage(), t);
+        if (null != callback) {
+          callback.failed(dsoe);
+        } else {
+          defaultErrorHandler.failed(dsoe);
+        }
+      }
+    }
+
+  }
+
   protected class AsyncPost<V> implements Runnable {
 
     private String bucket;
@@ -280,7 +367,10 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck
               entity));
         }
         if (null != callback) {
-          callback.completed(extractMetaData(result.getHeaders()), (V) result.getBody());
+          RiakMetaData meta = extractMetaData(result.getHeaders());
+          meta.setBucket((null != bucket ? bucket.toString() : null));
+          meta.setKey((null != key ? key.toString() : null));
+          callback.completed(meta, (V) result.getBody());
         }
       } catch (Throwable t) {
         DataStoreOperationException dsoe = new DataStoreOperationException(t.getMessage(), t);
@@ -316,6 +406,8 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck
             key);
         if (result.hasBody()) {
           RiakMetaData meta = extractMetaData(result.getHeaders());
+          meta.setBucket((null != bucket ? bucket.toString() : null));
+          meta.setKey((null != key ? key.toString() : null));
           RiakValue<T> val = new RiakValue<T>(result.getBody(), meta);
           if (useCache) {
             cache.put(new SimpleBucketKeyPair<Object, Object>(bucket, key), val);
diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/KeyValueStoreMetaData.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/KeyValueStoreMetaData.java
index 9a95c572c..4773811cd 100644
--- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/KeyValueStoreMetaData.java
+++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/KeyValueStoreMetaData.java
@@ -29,6 +29,10 @@ import java.util.Map;
  */
 public interface KeyValueStoreMetaData {
 
+  String getBucket();
+
+  String getKey();
+
   /**
    * Get the Content-Type of this object.
    *
diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/RiakMetaData.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/RiakMetaData.java
index 2c96eff21..12d80079c 100644
--- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/RiakMetaData.java
+++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/RiakMetaData.java
@@ -33,6 +33,8 @@ public class RiakMetaData implements KeyValueStoreMetaData {
 
   private MediaType mediaType = MediaType.APPLICATION_JSON;
   private Map<String, Object> properties;
+  private String bucket = null;
+  private String key = null;
 
   public RiakMetaData(Map<String, Object> properties) {
     this.properties = properties;
@@ -43,6 +45,29 @@ public class RiakMetaData implements KeyValueStoreMetaData {
     this.properties = properties;
   }
 
+  public RiakMetaData(MediaType mediaType, Map<String, Object> properties, String bucket, String key) {
+    this.mediaType = mediaType;
+    this.properties = properties;
+    this.bucket = bucket;
+    this.key = key;
+  }
+
+  public void setBucket(String bucket) {
+    this.bucket = bucket;
+  }
+
+  public void setKey(String key) {
+    this.key = key;
+  }
+
+  public String getBucket() {
+    return this.bucket;
+  }
+
+  public String getKey() {
+    return this.key;
+  }
+
   public MediaType getContentType() {
     return mediaType;
   }
diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/RiakTemplate.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/RiakTemplate.java
index 106c16c33..432789fb6 100644
--- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/RiakTemplate.java
+++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/RiakTemplate.java
@@ -211,7 +211,10 @@ public class RiakTemplate extends AbstractRiakTemplate implements BucketKeyValue
     HttpHeaders headers;
     try {
       headers = restTemplate.headForHeaders(defaultUri, bucket, key);
-      return extractMetaData(headers);
+      RiakMetaData meta = extractMetaData(headers);
+      meta.setBucket((null != bucket ? bucket.toString() : null));
+      meta.setKey((null != key ? key.toString() : null));
+      return meta;
     } catch (ResourceAccessException e) {
     } catch (IOException e) {
       throw new DataAccessResourceFailureException(e.getMessage(), e);
@@ -309,7 +312,7 @@ public class RiakTemplate extends AbstractRiakTemplate implements BucketKeyValue
   }
 
   @SuppressWarnings({"unchecked"})
-  public <B, K> RiakValue<byte[]> getAsBytesWithMetaData(B bucket, K key) {
+  public <B, K> RiakValue<byte[]> getAsBytesWithMetaData(final B bucket, final K key) {
     final RestTemplate restTemplate = getRestTemplate();
     if (log.isDebugEnabled()) {
       log.debug(String.format("GET object: bucket=%s, key=%s, type=byte[]",
@@ -343,6 +346,8 @@ public class RiakTemplate extends AbstractRiakTemplate implements BucketKeyValue
 
               HttpHeaders headers = response.getHeaders();
               RiakMetaData meta = extractMetaData(headers);
+              meta.setBucket((null != bucket ? bucket.toString() : null));
+              meta.setKey((null != key ? key.toString() : null));
               RiakValue<byte[]> val = new RiakValue<byte[]>(out.toByteArray(),
                   meta);
               return val;
diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakBuilder.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakBuilder.java
index 880fd29b9..6a37f0449 100644
--- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakBuilder.java
+++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakBuilder.java
@@ -150,11 +150,14 @@ public class RiakBuilder extends BuilderSupport {
     if (log.isDebugEnabled()) {
       log.debug("invokeMethod: " + methodName + " " + arg);
     }
+
+    Object[] args = (Object[]) arg;
+    Map<String, Object> params;
+    Closure handler = null;
+    RiakOperation<Object> op;
+
     if ("completed".equals(methodName) || "failed".equals(methodName)) {
-      RiakOperation<Object> op = (RiakOperation<Object>) getCurrent();
-      Object[] args = (Object[]) arg;
-      Map<String, Object> params;
-      Closure handler = null;
+      op = (RiakOperation<Object>) getCurrent();
       Closure guard = null;
       for (Object o : args) {
         if (o instanceof Map) {
@@ -169,6 +172,7 @@ public class RiakBuilder extends BuilderSupport {
       op.addHandler(methodName, handler, guard);
       return op;
     }
+
     return super.invokeMethod(methodName, arg);
   }
 
@@ -188,9 +192,11 @@ public class RiakBuilder extends BuilderSupport {
     }
   }
 
+  @SuppressWarnings({"unchecked"})
   @Override
   protected Object postNodeCompletion(Object parent, Object node) {
     log.debug("postNodeCompletion: " + parent + " " + node);
     return super.postNodeCompletion(parent, node);
   }
+
 }
diff --git a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakOperation.java b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakOperation.java
index 9d8b4d02a..d0563a96d 100644
--- a/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakOperation.java
+++ b/spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakOperation.java
@@ -21,6 +21,7 @@ package org.springframework.data.keyvalue.riak.groovy;
 import groovy.lang.Closure;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.data.keyvalue.riak.DataStoreOperationException;
 import org.springframework.data.keyvalue.riak.core.AsyncKeyValueStoreOperation;
 import org.springframework.data.keyvalue.riak.core.AsyncRiakTemplate;
 import org.springframework.data.keyvalue.riak.core.KeyValueStoreMetaData;
@@ -30,9 +31,7 @@ import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.Callable;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 
 /**
  * @author J. Brisbin <jon@jbrisbin.com>
@@ -40,7 +39,7 @@ import java.util.concurrent.TimeUnit;
 public class RiakOperation<T> implements Callable {
 
   static enum Type {
-    SET, SETASBYTES, PUT, GET, GETASBYTES, CONTAINSKEY, DELETE
+    SET, SETASBYTES, PUT, GET, GETASBYTES, CONTAINSKEY, DELETE, EACH
   }
 
   static String COMPLETED = "completed";
@@ -131,16 +130,19 @@ public class RiakOperation<T> implements Callable {
         f = riak.getAsBytes(bucket, key, callbackInvoker);
         break;
       case PUT:
-        throw new IllegalStateException("PUT not yet implemented in AsyncRiakTemplate");
+        f = riak.put(bucket, value, callbackInvoker);
+        break;
       case SET:
         f = riak.set(bucket, key, value, callbackInvoker);
         break;
       case SETASBYTES:
+        byte[] bytes;
         if (value instanceof byte[]) {
-          f = riak.setAsBytes(bucket, key, (byte[]) value, callbackInvoker);
+          bytes = (byte[]) value;
         } else {
-          log.error("Need to convert obj to byte array first!");
+          bytes = riak.getConversionService().convert(value, byte[].class);
         }
+        f = riak.setAsBytes(bucket, key, bytes, callbackInvoker);
         break;
       case CONTAINSKEY:
         f = riak.containsKey(bucket, key, callbackInvoker);
@@ -148,6 +150,35 @@ public class RiakOperation<T> implements Callable {
       case DELETE:
         f = riak.delete(bucket, key, callbackInvoker);
         break;
+      case EACH:
+        f = riak.getBucketSchema(bucket,
+            null,
+            new AsyncKeyValueStoreOperation<Map<String, Object>>() {
+              public void completed(KeyValueStoreMetaData meta, Map<String, Object> result) {
+                List<String> keys = (List<String>) result.get("keys");
+                for (String key : keys) {
+                  try {
+                    Future<?> getFut = riak.get(bucket, key, callbackInvoker);
+                    if (timeout > 0) {
+                      getFut.get(timeout, TimeUnit.MILLISECONDS);
+                    } else if (timeout < 0) {
+                      getFut.get();
+                    }
+                  } catch (InterruptedException e) {
+                    throw new DataStoreOperationException(e.getMessage(), e);
+                  } catch (ExecutionException e) {
+                    throw new DataStoreOperationException(e.getMessage(), e);
+                  } catch (TimeoutException e) {
+                    throw new DataStoreOperationException(e.getMessage(), e);
+                  }
+                }
+              }
+
+              public void failed(Throwable error) {
+                log.error(error.getMessage(), error);
+              }
+            });
+        break;
     }
 
     if (null != f) {
@@ -189,6 +220,9 @@ public class RiakOperation<T> implements Callable {
   class ClosureInvokingCallback implements AsyncKeyValueStoreOperation {
 
     public void completed(KeyValueStoreMetaData meta, Object result) {
+      if (!callbacks.containsKey(COMPLETED)) {
+        return;
+      }
       for (GuardedClosure cl : callbacks.get(COMPLETED)) {
         boolean execute = true;
 
diff --git a/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakBuilderSpec.groovy b/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakBuilderSpec.groovy
index 981369927..c47c0d0f2 100644
--- a/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakBuilderSpec.groovy
+++ b/spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakBuilderSpec.groovy
@@ -119,21 +119,61 @@ class RiakBuilderSpec extends Specification {
 
   }
 
-  def "Test builder delete"() {
+  def "Test builder put"() {
+
+    given:
+    def obj = [test: "value", integer: 12]
+    def riak = new RiakBuilder(riakTemplate)
+    def id = null
+
+    when:
+    riak.put(bucket: "test", qos: [dw: "all"], value: obj) {
+
+      completed { v, meta ->
+        id = meta.key
+      }
+
+      failed { e -> println "failure: $e" }
+
+    }
+
+    then:
+    null != id
+
+  }
+
+  def "Test builder each"() {
 
     given:
     def riak = new RiakBuilder(riakTemplate)
-    def result = null
+    def idCnt = 0
 
     when:
-    riak.delete(bucket: "test", key: "test", wait: 3000L) {
-      completed { v -> result = v }
+    riak.each(bucket: "test") {
+      completed { v, meta -> idCnt++ }
       failed { e -> println "failure: $e" }
     }
 
     then:
-    null != result
-    result
+    idCnt > 0
+
+  }
+
+  def "Test builder delete"() {
+
+    given:
+    def riak = new RiakBuilder(riakTemplate)
+
+    when:
+    riak.each(bucket: "test") {
+      completed { v, meta ->
+        delete(bucket: meta.bucket, key: meta.key)
+      }
+      failed { e -> println "failure: $e" }
+    }
+
+    then:
+    true
 
   }