diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosContainerProperties.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosContainerProperties.java
index c6d2d22fb88c..695d1bb38694 100644
--- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosContainerProperties.java
+++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosContainerProperties.java
@@ -196,6 +196,46 @@ public CosmosContainerProperties setConflictResolutionPolicy(ConflictResolutionP
         return this;
     }
 
+    /**
+     * Gets the collection's default time-to-live value.
+     *
+     * @return the default time-to-live value in seconds.
+     */
+    public Integer getDefaultTimeToLiveInSeconds() {
+        if (super.has(Constants.Properties.DEFAULT_TTL)) {
+            return super.getInt(Constants.Properties.DEFAULT_TTL);
+        }
+
+        return null;
+    }
+
+    /**
+     * Sets the collection's default time-to-live value.
+     * <p>
+     * The default time-to-live value on a collection is an optional property. If set, the documents within the collection
+     * expires after the specified number of seconds since their last write time. The value of this property should be one of the following:
+     * <p>
+     * null - indicates evaluation of time-to-live is disabled and documents within the collection will never expire, regardless whether
+     * individual documents have their time-to-live set.
+     * <p>
+     * nonzero positive integer - indicates the default time-to-live value for all documents within the collection. This value can be overridden
+     * by individual documents' time-to-live value.
+     * <p>
+     * -1 - indicates by default all documents within the collection never expire. This value can be overridden by individual documents'
+     * time-to-live value.
+     *
+     * @param timeToLive the default time-to-live value in seconds.
+     */
+    public void setDefaultTimeToLiveInSeconds(Integer timeToLive) {
+        // a "null" value is represented as a missing element on the wire.
+        // setting timeToLive to null should remove the property from the property bag.
+        if (timeToLive != null) {
+            super.set(Constants.Properties.DEFAULT_TTL, timeToLive);
+        } else if (super.has(Constants.Properties.DEFAULT_TTL)) {
+            super.remove(Constants.Properties.DEFAULT_TTL);
+        }
+    }
+
     DocumentCollection getV2Collection() {
         DocumentCollection collection = new DocumentCollection(this.toJson());
         collection.setPartitionKey(this.getPartitionKeyDefinition());
diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosResponseValidator.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosResponseValidator.java
index cefed5a6f9a2..f5af62385c78 100644
--- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosResponseValidator.java
+++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosResponseValidator.java
@@ -88,6 +88,19 @@ public void validate(CosmosAsyncContainerResponse resourceResponse) {
             return this;
         }
 
+        public Builder<T> withDefaultTimeToLive(Integer timeToLive) {
+            validators.add(new CosmosResponseValidator<CosmosAsyncContainerResponse>() {
+
+                @Override
+                public void validate(CosmosAsyncContainerResponse resourceResponse) {
+                    assertThat(resourceResponse.getProperties()).isNotNull();
+                    assertThat(resourceResponse.getProperties().getDefaultTimeToLiveInSeconds()).isNotNull();
+                    assertThat(resourceResponse.getProperties().getDefaultTimeToLiveInSeconds()).isEqualTo(timeToLive);
+                }
+            });
+            return this;
+        }
+
         public Builder<T> withProperty(String propertyName, String value) {
             validators.add(new CosmosResponseValidator<T>() {
                 @Override
diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/CollectionCrudTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/CollectionCrudTest.java
index 28c4b7e48d10..709e7250daf9 100644
--- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/CollectionCrudTest.java
+++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/CollectionCrudTest.java
@@ -95,6 +95,23 @@ public void createCollection(String collectionName) throws InterruptedException
         safeDeleteAllCollections(database);
     }
 
+    @Test(groups = { "emulator" }, timeOut = TIMEOUT, dataProvider = "collectionCrudArgProvider")
+    public void createCollectionWithTTL(String collectionName) throws InterruptedException {
+        CosmosContainerProperties collectionDefinition = getCollectionDefinition(collectionName);
+
+        Integer defaultTimeToLive = 300;
+        collectionDefinition.setDefaultTimeToLiveInSeconds(defaultTimeToLive);
+
+        Mono<CosmosAsyncContainerResponse> createObservable = database
+            .createContainer(collectionDefinition);
+
+        CosmosResponseValidator<CosmosAsyncContainerResponse> validator = new CosmosResponseValidator.Builder<CosmosAsyncContainerResponse>()
+            .withId(collectionDefinition.getId()).withDefaultTimeToLive(defaultTimeToLive).build();
+
+        validateSuccess(createObservable, validator);
+        safeDeleteAllCollections(database);
+    }
+
     @Test(groups = { "emulator" }, timeOut = TIMEOUT)
     public void createCollectionWithCompositeIndexAndSpatialSpec() throws InterruptedException {
         PartitionKeyDefinition partitionKeyDef = new PartitionKeyDefinition();
@@ -188,6 +205,24 @@ public void readCollection(String collectionName) throws InterruptedException {
         safeDeleteAllCollections(database);
     }
 
+    @Test(groups = { "emulator" }, timeOut = TIMEOUT, dataProvider = "collectionCrudArgProvider")
+    public void readCollectionWithTTL(String collectionName) throws InterruptedException {
+        CosmosContainerProperties collectionDefinition = getCollectionDefinition(collectionName);
+
+        Integer defaultTimeToLive = 200;
+        collectionDefinition.setDefaultTimeToLiveInSeconds(defaultTimeToLive);
+
+        Mono<CosmosAsyncContainerResponse> createObservable = database.createContainer(collectionDefinition);
+        CosmosAsyncContainer collection = createObservable.block().getContainer();
+
+        Mono<CosmosAsyncContainerResponse> readObservable = collection.read();
+
+        CosmosResponseValidator<CosmosAsyncContainerResponse> validator = new CosmosResponseValidator.Builder<CosmosAsyncContainerResponse>()
+            .withId(collection.getId()).withDefaultTimeToLive(defaultTimeToLive).build();
+        validateSuccess(readObservable, validator);
+        safeDeleteAllCollections(database);
+    }
+
     @Test(groups = { "emulator" }, timeOut = TIMEOUT, dataProvider = "collectionCrudArgProvider")
     public void readCollection_DoesntExist(String collectionName) throws Exception {
 
@@ -235,6 +270,34 @@ public void replaceCollection(String collectionName) throws InterruptedException
         safeDeleteAllCollections(database);
     }
 
+    @Test(groups = { "emulator" }, timeOut = TIMEOUT, dataProvider = "collectionCrudArgProvider")
+    public void replaceCollectionWithTTL(String collectionName) throws InterruptedException {
+        // create a collection
+        CosmosContainerProperties collectionDefinition = getCollectionDefinition(collectionName);
+        Integer defaultTimeToLive = 120;
+        collectionDefinition.setDefaultTimeToLiveInSeconds(defaultTimeToLive);
+        Mono<CosmosAsyncContainerResponse> createObservable = database.createContainer(collectionDefinition);
+        CosmosAsyncContainer collection = createObservable.block().getContainer();
+        CosmosContainerProperties collectionSettings = collection.read().block().getProperties();
+        // sanity check
+        assertThat(collectionSettings.getIndexingPolicy().getIndexingMode()).isEqualTo(IndexingMode.CONSISTENT);
+        assertThat(collectionSettings.getDefaultTimeToLiveInSeconds()).isEqualTo(defaultTimeToLive);
+
+        // replace indexing mode
+        IndexingPolicy indexingMode = new IndexingPolicy();
+        indexingMode.setIndexingMode(IndexingMode.LAZY);
+        collectionSettings.setIndexingPolicy(indexingMode);
+        collectionSettings.setDefaultTimeToLiveInSeconds(defaultTimeToLive * 2);
+        Mono<CosmosAsyncContainerResponse> readObservable = collection.replace(collectionSettings, new CosmosContainerRequestOptions());
+
+        // validate
+        CosmosResponseValidator<CosmosAsyncContainerResponse> validator = new CosmosResponseValidator.Builder<CosmosAsyncContainerResponse>()
+            .indexingMode(IndexingMode.LAZY).withDefaultTimeToLive(defaultTimeToLive * 2).build();
+        validateSuccess(readObservable, validator);
+        safeDeleteAllCollections(database);
+    }
+
+
     @Test(groups = { "emulator" }, timeOut = 10 * TIMEOUT, retryAnalyzer = RetryAnalyzer.class)
     public void sessionTokenConsistencyCollectionDeleteCreateSameName() {
         CosmosAsyncClient client1 = clientBuilder().buildAsyncClient();