diff --git a/syncbase/src/main/java/io/v/syncbase/Database.java b/syncbase/src/main/java/io/v/syncbase/Database.java
index fc67c83..00c0858 100644
--- a/syncbase/src/main/java/io/v/syncbase/Database.java
+++ b/syncbase/src/main/java/io/v/syncbase/Database.java
@@ -392,8 +392,72 @@
      * Options for {@code addWatchChangeHandler}.
      */
     public static class AddWatchChangeHandlerOptions {
-        public byte[] resumeMarker;
-        boolean showUserdataCollectionRow;
+        final byte[] resumeMarker;
+        final String name;
+        final String blessing;
+        final String row;
+        final boolean showUserdataCollectionRow;
+
+        AddWatchChangeHandlerOptions(Builder builder) {
+            resumeMarker = builder.resumeMarker;
+            name = builder.name;
+            blessing = builder.blessing;
+            row = builder.row;
+            showUserdataCollectionRow = builder.showUserdataCollectionRow;
+        }
+
+        CollectionRowPattern getCollectionRowPattern() {
+            return new CollectionRowPattern(blessing, name, row);
+        }
+
+        public static class Builder {
+            private byte[] resumeMarker;
+            private String name = WILDCARD;
+            private String blessing = WILDCARD;
+            private String row = WILDCARD;
+            private boolean showUserdataCollectionRow;
+
+            private static final String WILDCARD = "%";
+
+            public Builder setResumeMarker(byte[] resumeMarker) {
+                this.resumeMarker = resumeMarker;
+                return this;
+            }
+
+            public Builder setCollectionNamePrefix(String prefix) {
+                // TODO(alexfandrianto): Unsafe. The prefix was not escaped. Incorrect if it has a
+                // trailing backslash.
+                name = prefix + WILDCARD;
+                return this;
+            }
+
+            public Builder setCollectionId(Id id) {
+                name = id.getName();
+                blessing = id.getBlessing();
+                return this;
+            }
+
+            public Builder setRowKeyPrefix(String prefix) {
+                // TODO(alexfandrianto): Unsafe. The prefix was not escaped. Incorrect if it has a
+                // trailing backslash.
+                row = prefix + WILDCARD;
+                return this;
+            }
+
+            public Builder setRowKey(String rowKey) {
+                row = rowKey;
+                return this;
+            }
+
+            Builder setShowUserdataCollectionRow(boolean shouldShow) {
+                showUserdataCollectionRow = shouldShow;
+                return this;
+            }
+
+            public AddWatchChangeHandlerOptions build() {
+                return new AddWatchChangeHandlerOptions(this);
+            }
+        }
     }
 
     /**
@@ -438,7 +502,7 @@
             throw new UnsupportedOperationException("Specifying resumeMarker is not yet supported");
         }
 
-        mCoreDatabase.watch(null, ImmutableList.of(new CollectionRowPattern("%", "%", "%")),
+        mCoreDatabase.watch(null, ImmutableList.of(opts.getCollectionRowPattern()),
                 new io.v.syncbase.core.Database.WatchPatternsCallbacks() {
                     private boolean mGotFirstBatch = false;
                     private final List<WatchChange> mBatch = new ArrayList<>();
@@ -509,7 +573,7 @@
      * {@code AddWatchChangeHandlerOptions}.
      */
     public void addWatchChangeHandler(WatchChangeHandler h) {
-        addWatchChangeHandler(h, new AddWatchChangeHandlerOptions());
+        addWatchChangeHandler(h, new AddWatchChangeHandlerOptions.Builder().build());
     }
 
     /**
diff --git a/syncbase/src/main/java/io/v/syncbase/Syncbase.java b/syncbase/src/main/java/io/v/syncbase/Syncbase.java
index e026642..56083da 100644
--- a/syncbase/src/main/java/io/v/syncbase/Syncbase.java
+++ b/syncbase/src/main/java/io/v/syncbase/Syncbase.java
@@ -267,8 +267,8 @@
                             syncgroup.createIfMissing(ImmutableList.of(sUserdataCollection));
                         }
                         Database.AddWatchChangeHandlerOptions opts = new Database
-                                .AddWatchChangeHandlerOptions();
-                        opts.showUserdataCollectionRow = true;
+                                .AddWatchChangeHandlerOptions.Builder().
+                                setShowUserdataCollectionRow(true).build();
                         sDatabase.addWatchChangeHandler(new UserdataWatchHandler(), opts);
                     }
                     sOpts.callbackExecutor.execute(new Runnable() {
diff --git a/syncbase/src/test/java/io/v/syncbase/SyncbaseTest.java b/syncbase/src/test/java/io/v/syncbase/SyncbaseTest.java
index f8fd471..8513424 100644
--- a/syncbase/src/test/java/io/v/syncbase/SyncbaseTest.java
+++ b/syncbase/src/test/java/io/v/syncbase/SyncbaseTest.java
@@ -215,8 +215,8 @@
                 WatchChange watchChange = changes.next();
                 assertEquals(WatchChange.ChangeType.DELETE, watchChange.getChangeType());
                 assertTrue(watchChange.getCollectionId().getName().startsWith("c"));
-                // TODO(razvanm): Uncomment after the POJO start working.
-                //assertEquals(1, watchChange.getValue());
+                assertEquals("foo", watchChange.getRowKey());
+
                 assertFalse(changes.hasNext());
                 waitOnChangeBatch.set(null);
             }
@@ -235,6 +235,140 @@
     }
 
     @Test
+    public void testWatchSpecificCollection() throws Exception {
+        Database db = createDatabase();
+        final SettableFuture<Void> waitOnInitialState = SettableFuture.create();
+        final SettableFuture<Void> waitOnChangeBatch = SettableFuture.create();
+        Collection collection = db.createCollection();
+        collection.put("foo", 1);
+        // Note: For ease of test writing, "gar" is a key lexicographically after "foo".
+        collection.put("gar", 2);
+        final String collectionName = collection.getId().getName();
+        Database.AddWatchChangeHandlerOptions opts = new Database.AddWatchChangeHandlerOptions.
+                Builder().setCollectionId(collection.getId()).build();
+        db.addWatchChangeHandler(new Database.WatchChangeHandler() {
+            @Override
+            public void onInitialState(Iterator<WatchChange> values) {
+                // TODO(razvanm): Check the entire contents of each change.
+                // 1st change: the collection entity for the "c" collection.
+                assertTrue(values.hasNext());
+                WatchChange watchChange = (WatchChange) values.next();
+                assertEquals(WatchChange.EntityType.COLLECTION, watchChange.getEntityType());
+                assertEquals(WatchChange.ChangeType.PUT, watchChange.getChangeType());
+                assertTrue(watchChange.getCollectionId().getName().equals(collectionName));
+                // 2nd change: the row for the "foo" key.
+                assertTrue(values.hasNext());
+                watchChange = (WatchChange) values.next();
+                assertEquals(WatchChange.EntityType.ROW, watchChange.getEntityType());
+                assertEquals(WatchChange.ChangeType.PUT, watchChange.getChangeType());
+                assertTrue(watchChange.getCollectionId().getName().equals(collectionName));
+                assertEquals("foo", watchChange.getRowKey());
+                // TODO(razvanm): Uncomment after the POJO start working.
+                //assertEquals(1, watchChange.getValue());
+                // 3rd change: the row for the "gar" key.
+                assertTrue(values.hasNext());
+                watchChange = (WatchChange) values.next();
+                assertEquals(WatchChange.EntityType.ROW, watchChange.getEntityType());
+                assertEquals(WatchChange.ChangeType.PUT, watchChange.getChangeType());
+                assertTrue(watchChange.getCollectionId().getName().equals(collectionName));
+                assertEquals("gar", watchChange.getRowKey());
+                // TODO(razvanm): Uncomment after the POJO start working.
+                //assertEquals(2, watchChange.getValue());
+
+                // No more changes.
+                assertFalse(values.hasNext());
+                waitOnInitialState.set(null);
+            }
+
+            @Override
+            public void onChangeBatch(Iterator<WatchChange> changes) {
+                assertTrue(changes.hasNext());
+                WatchChange watchChange = changes.next();
+                assertEquals(WatchChange.ChangeType.DELETE, watchChange.getChangeType());
+                assertTrue(watchChange.getCollectionId().getName().startsWith("c"));
+                assertEquals("foo", watchChange.getRowKey());
+
+                assertFalse(changes.hasNext());
+                waitOnChangeBatch.set(null);
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                VError vError = (VError) e;
+                assertEquals("v.io/v23/verror.Unknown", vError.id);
+                assertEquals("context canceled", vError.message);
+                assertEquals(0, vError.actionCode);
+            }
+        }, opts);
+        waitOnInitialState.get(1, TimeUnit.SECONDS);
+        collection.delete("foo");
+        waitOnChangeBatch.get(1, TimeUnit.SECONDS);
+    }
+
+
+    @Test
+    public void testWatchSpecificCollectionWithRowPrefix() throws Exception {
+        Database db = createDatabase();
+        final SettableFuture<Void> waitOnInitialState = SettableFuture.create();
+        final SettableFuture<Void> waitOnChangeBatch = SettableFuture.create();
+        Collection collection = db.createCollection();
+        collection.put("foo", 1);
+        collection.put("bar", 1); // not seen, due to filter
+        final String collectionName = collection.getId().getName();
+        Database.AddWatchChangeHandlerOptions opts = new Database.AddWatchChangeHandlerOptions.
+                Builder().setCollectionId(collection.getId()).setRowKeyPrefix("f").build();
+        db.addWatchChangeHandler(new Database.WatchChangeHandler() {
+            @Override
+            public void onInitialState(Iterator<WatchChange> values) {
+                // TODO(razvanm): Check the entire contents of each change.
+                // 1st change: the collection entity for the "c" collection.
+                assertTrue(values.hasNext());
+                WatchChange watchChange = (WatchChange) values.next();
+                assertEquals(WatchChange.EntityType.COLLECTION, watchChange.getEntityType());
+                assertEquals(WatchChange.ChangeType.PUT, watchChange.getChangeType());
+                assertTrue(watchChange.getCollectionId().getName().equals(collectionName));
+                // 2nd change: the row for the "foo" key.
+                assertTrue(values.hasNext());
+                watchChange = (WatchChange) values.next();
+                assertEquals(WatchChange.EntityType.ROW, watchChange.getEntityType());
+                assertEquals(WatchChange.ChangeType.PUT, watchChange.getChangeType());
+                assertTrue(watchChange.getCollectionId().getName().equals(collectionName));
+                assertEquals("foo", watchChange.getRowKey());
+                // TODO(razvanm): Uncomment after the POJO start working.
+                //assertEquals(1, watchChange.getValue());
+
+                // No more changes.
+                assertFalse(values.hasNext());
+                waitOnInitialState.set(null);
+            }
+
+            @Override
+            public void onChangeBatch(Iterator<WatchChange> changes) {
+                assertTrue(changes.hasNext());
+                WatchChange watchChange = changes.next();
+                assertEquals(WatchChange.ChangeType.DELETE, watchChange.getChangeType());
+                assertTrue(watchChange.getCollectionId().getName().startsWith("c"));
+                assertEquals("foo", watchChange.getRowKey());
+
+                assertFalse(changes.hasNext());
+                waitOnChangeBatch.set(null);
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                VError vError = (VError) e;
+                assertEquals("v.io/v23/verror.Unknown", vError.id);
+                assertEquals("context canceled", vError.message);
+                assertEquals(0, vError.actionCode);
+            }
+        }, opts);
+        waitOnInitialState.get(1, TimeUnit.SECONDS);
+        collection.delete("foo");
+        collection.delete("bar"); // not noticed due to filter.
+        waitOnChangeBatch.get(1, TimeUnit.SECONDS);
+    }
+
+    @Test
     public void testRunInBatch() throws Exception {
         Database db = createDatabase();