diff --git a/builtin/providers/aws/resource_aws_elasticache_cluster.go b/builtin/providers/aws/resource_aws_elasticache_cluster.go
index 700b7ddc8b9e..3460fb292ff8 100644
--- a/builtin/providers/aws/resource_aws_elasticache_cluster.go
+++ b/builtin/providers/aws/resource_aws_elasticache_cluster.go
@@ -118,7 +118,10 @@ func resourceAwsElasticacheCluster() *schema.Resource {
 					},
 				},
 			},
-
+			"notification_topic_arn": &schema.Schema{
+				Type:     schema.TypeString,
+				Optional: true,
+			},
 			// A single-element string list containing an Amazon Resource Name (ARN) that
 			// uniquely identifies a Redis RDB snapshot file stored in Amazon S3. The snapshot
 			// file will be used to populate the node group.
@@ -188,6 +191,10 @@ func resourceAwsElasticacheClusterCreate(d *schema.ResourceData, meta interface{
 		req.PreferredMaintenanceWindow = aws.String(v.(string))
 	}
 
+	if v, ok := d.GetOk("notification_topic_arn"); ok {
+		req.NotificationTopicArn = aws.String(v.(string))
+	}
+
 	snaps := d.Get("snapshot_arns").(*schema.Set).List()
 	if len(snaps) > 0 {
 		s := expandStringList(snaps)
@@ -254,6 +261,11 @@ func resourceAwsElasticacheClusterRead(d *schema.ResourceData, meta interface{})
 		d.Set("security_group_ids", c.SecurityGroups)
 		d.Set("parameter_group_name", c.CacheParameterGroup)
 		d.Set("maintenance_window", c.PreferredMaintenanceWindow)
+		if c.NotificationConfiguration != nil {
+			if *c.NotificationConfiguration.TopicStatus == "active" {
+				d.Set("notification_topic_arn", c.NotificationConfiguration.TopicArn)
+			}
+		}
 
 		if err := setCacheNodeData(d, c); err != nil {
 			return err
@@ -317,6 +329,16 @@ func resourceAwsElasticacheClusterUpdate(d *schema.ResourceData, meta interface{
 		requestUpdate = true
 	}
 
+	if d.HasChange("notification_topic_arn") {
+		v := d.Get("notification_topic_arn").(string)
+		req.NotificationTopicArn = aws.String(v)
+		if v == "" {
+			inactive := "inactive"
+			req.NotificationTopicStatus = &inactive
+		}
+		requestUpdate = true
+	}
+
 	if d.HasChange("engine_version") {
 		req.EngineVersion = aws.String(d.Get("engine_version").(string))
 		requestUpdate = true
diff --git a/builtin/providers/aws/resource_aws_elasticache_cluster_test.go b/builtin/providers/aws/resource_aws_elasticache_cluster_test.go
index 173ca21ea761..b9306002886a 100644
--- a/builtin/providers/aws/resource_aws_elasticache_cluster_test.go
+++ b/builtin/providers/aws/resource_aws_elasticache_cluster_test.go
@@ -3,6 +3,7 @@ package aws
 import (
 	"fmt"
 	"math/rand"
+	"strings"
 	"testing"
 	"time"
 
@@ -13,6 +14,7 @@ import (
 )
 
 func TestAccAWSElasticacheCluster_basic(t *testing.T) {
+	var ec elasticache.CacheCluster
 	resource.Test(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
 		Providers:    testAccProviders,
@@ -22,7 +24,7 @@ func TestAccAWSElasticacheCluster_basic(t *testing.T) {
 				Config: testAccAWSElasticacheClusterConfig,
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckAWSElasticacheSecurityGroupExists("aws_elasticache_security_group.bar"),
-					testAccCheckAWSElasticacheClusterExists("aws_elasticache_cluster.bar"),
+					testAccCheckAWSElasticacheClusterExists("aws_elasticache_cluster.bar", &ec),
 					resource.TestCheckResourceAttr(
 						"aws_elasticache_cluster.bar", "cache_nodes.0.id", "0001"),
 				),
@@ -33,6 +35,7 @@ func TestAccAWSElasticacheCluster_basic(t *testing.T) {
 
 func TestAccAWSElasticacheCluster_vpc(t *testing.T) {
 	var csg elasticache.CacheSubnetGroup
+	var ec elasticache.CacheCluster
 	resource.Test(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
 		Providers:    testAccProviders,
@@ -42,13 +45,28 @@ func TestAccAWSElasticacheCluster_vpc(t *testing.T) {
 				Config: testAccAWSElasticacheClusterInVPCConfig,
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckAWSElasticacheSubnetGroupExists("aws_elasticache_subnet_group.bar", &csg),
-					testAccCheckAWSElasticacheClusterExists("aws_elasticache_cluster.bar"),
+					testAccCheckAWSElasticacheClusterExists("aws_elasticache_cluster.bar", &ec),
+					testAccCheckAWSElasticacheClusterAttributes(&ec),
 				),
 			},
 		},
 	})
 }
 
+func testAccCheckAWSElasticacheClusterAttributes(v *elasticache.CacheCluster) resource.TestCheckFunc {
+	return func(s *terraform.State) error {
+		if v.NotificationConfiguration == nil {
+			return fmt.Errorf("Expected NotificationConfiguration for ElastiCache Cluster (%s)", *v.CacheClusterId)
+		}
+
+		if strings.ToLower(*v.NotificationConfiguration.TopicStatus) != "active" {
+			return fmt.Errorf("Expected NotificationConfiguration status to be 'active', got (%s)", *v.NotificationConfiguration.TopicStatus)
+		}
+
+		return nil
+	}
+}
+
 func testAccCheckAWSElasticacheClusterDestroy(s *terraform.State) error {
 	conn := testAccProvider.Meta().(*AWSClient).elasticacheconn
 
@@ -69,7 +87,7 @@ func testAccCheckAWSElasticacheClusterDestroy(s *terraform.State) error {
 	return nil
 }
 
-func testAccCheckAWSElasticacheClusterExists(n string) resource.TestCheckFunc {
+func testAccCheckAWSElasticacheClusterExists(n string, v *elasticache.CacheCluster) resource.TestCheckFunc {
 	return func(s *terraform.State) error {
 		rs, ok := s.RootModule().Resources[n]
 		if !ok {
@@ -81,12 +99,19 @@ func testAccCheckAWSElasticacheClusterExists(n string) resource.TestCheckFunc {
 		}
 
 		conn := testAccProvider.Meta().(*AWSClient).elasticacheconn
-		_, err := conn.DescribeCacheClusters(&elasticache.DescribeCacheClustersInput{
+		resp, err := conn.DescribeCacheClusters(&elasticache.DescribeCacheClustersInput{
 			CacheClusterId: aws.String(rs.Primary.ID),
 		})
 		if err != nil {
 			return fmt.Errorf("Elasticache error: %v", err)
 		}
+
+		for _, c := range resp.CacheClusters {
+			if *c.CacheClusterId == rs.Primary.ID {
+				*v = *c
+			}
+		}
+
 		return nil
 	}
 }
@@ -175,5 +200,10 @@ resource "aws_elasticache_cluster" "bar" {
     subnet_group_name = "${aws_elasticache_subnet_group.bar.name}"
     security_group_ids = ["${aws_security_group.bar.id}"]
     parameter_group_name = "default.redis2.8"
+    notification_topic_arn      = "${aws_sns_topic.topic_example.arn}"
+}
+
+resource "aws_sns_topic" "topic_example" {
+  name = "tf-ecache-cluster-test"
 }
 `, genRandInt(), genRandInt(), genRandInt())
diff --git a/website/source/docs/providers/aws/r/elasticache_cluster.html.markdown b/website/source/docs/providers/aws/r/elasticache_cluster.html.markdown
index f2d0e5363efd..ef1d69ed4a33 100644
--- a/website/source/docs/providers/aws/r/elasticache_cluster.html.markdown
+++ b/website/source/docs/providers/aws/r/elasticache_cluster.html.markdown
@@ -73,6 +73,10 @@ names to associate with this cache cluster
 Amazon Resource Name (ARN) of a Redis RDB snapshot file stored in Amazon S3. 
 Example: `arn:aws:s3:::my_bucket/snapshot1.rdb`
 
+* `notification_topic_arn` – (Optional) An Amazon Resource Name (ARN) of an 
+SNS topic to send ElastiCache notifications to. Example: 
+`arn:aws:sns:us-east-1:012345678999:my_sns_topic`
+
 * `tags` - (Optional) A mapping of tags to assign to the resource.