diff --git a/config/config.go b/config/config.go
index 055a7f3306f6..1772fd7e3639 100644
--- a/config/config.go
+++ b/config/config.go
@@ -231,7 +231,10 @@ func (r *Resource) Count() (int, error) {
 
 	v, err := strconv.ParseInt(count, 0, 0)
 	if err != nil {
-		return 0, err
+		return 0, fmt.Errorf(
+			"cannot parse %q as an integer",
+			count,
+		)
 	}
 
 	return int(v), nil
diff --git a/terraform/context_apply_test.go b/terraform/context_apply_test.go
index ddaaae0b4f9f..f620a43d7b3c 100644
--- a/terraform/context_apply_test.go
+++ b/terraform/context_apply_test.go
@@ -2421,6 +2421,73 @@ func TestContext2Apply_countVariableRef(t *testing.T) {
 	}
 }
 
+func TestContext2Apply_provisionerInterpCount(t *testing.T) {
+	// This test ensures that a provisioner can interpolate a resource count
+	// even though the provisioner expression is evaluated during the plan
+	// walk. https://github.com/hashicorp/terraform/issues/16840
+
+	m := testModule(t, "apply-provisioner-interp-count")
+
+	p := testProvider("aws")
+	p.ApplyFn = testApplyFn
+	p.DiffFn = testDiffFn
+
+	pr := testProvisioner()
+
+	providerResolver := ResourceProviderResolverFixed(
+		map[string]ResourceProviderFactory{
+			"aws": testProviderFuncFixed(p),
+		},
+	)
+	provisioners := map[string]ResourceProvisionerFactory{
+		"local-exec": testProvisionerFuncFixed(pr),
+	}
+	ctx := testContext2(t, &ContextOpts{
+		Module:           m,
+		ProviderResolver: providerResolver,
+		Provisioners:     provisioners,
+	})
+
+	plan, err := ctx.Plan()
+	if err != nil {
+		t.Fatalf("plan failed: %s", err)
+	}
+
+	// We'll marshal and unmarshal the plan here, to ensure that we have
+	// a clean new context as would be created if we separately ran
+	// terraform plan -out=tfplan && terraform apply tfplan
+	var planBuf bytes.Buffer
+	err = WritePlan(plan, &planBuf)
+	if err != nil {
+		t.Fatalf("failed to write plan: %s", err)
+	}
+	plan, err = ReadPlan(&planBuf)
+	if err != nil {
+		t.Fatalf("failed to read plan: %s", err)
+	}
+
+	ctx, err = plan.Context(&ContextOpts{
+		// Most options are taken from the plan in this case, but we still
+		// need to provide the plugins.
+		ProviderResolver: providerResolver,
+		Provisioners:     provisioners,
+	})
+	if err != nil {
+		t.Fatalf("failed to create context for plan: %s", err)
+	}
+
+	// Applying the plan should now succeed
+	_, err = ctx.Apply()
+	if err != nil {
+		t.Fatalf("apply failed: %s", err)
+	}
+
+	// Verify apply was invoked
+	if !pr.ApplyCalled {
+		t.Fatalf("provisioner was not applied")
+	}
+}
+
 func TestContext2Apply_mapVariableOverride(t *testing.T) {
 	m := testModule(t, "apply-map-var-override")
 	p := testProvider("aws")
diff --git a/terraform/node_resource_apply.go b/terraform/node_resource_apply.go
index 9f6d69fd3c79..90448065596b 100644
--- a/terraform/node_resource_apply.go
+++ b/terraform/node_resource_apply.go
@@ -124,6 +124,24 @@ func (n *NodeApplyableResource) evalTreeDataResource(
 				Then: EvalNoop{},
 			},
 
+			// Normally we interpolate count as a preparation step before
+			// a DynamicExpand, but an apply graph has pre-expanded nodes
+			// and so the count would otherwise never be interpolated.
+			//
+			// This is redundant when there are multiple instances created
+			// from the same config (count > 1) but harmless since the
+			// underlying structures have mutexes to make this concurrency-safe.
+			//
+			// In most cases this isn't actually needed because we dealt with
+			// all of the counts during the plan walk, but we do it here
+			// for completeness because other code assumes that the
+			// final count is always available during interpolation.
+			//
+			// Here we are just populating the interpolated value in-place
+			// inside this RawConfig object, like we would in
+			// NodeAbstractCountResource.
+			&EvalInterpolate{Config: n.Config.RawCount},
+
 			// We need to re-interpolate the config here, rather than
 			// just using the diff's values directly, because we've
 			// potentially learned more variable values during the
@@ -236,6 +254,25 @@ func (n *NodeApplyableResource) evalTreeManagedResource(
 				},
 			},
 
+			// Normally we interpolate count as a preparation step before
+			// a DynamicExpand, but an apply graph has pre-expanded nodes
+			// and so the count would otherwise never be interpolated.
+			//
+			// This is redundant when there are multiple instances created
+			// from the same config (count > 1) but harmless since the
+			// underlying structures have mutexes to make this concurrency-safe.
+			//
+			// In most cases this isn't actually needed because we dealt with
+			// all of the counts during the plan walk, but we need to do this
+			// in order to support interpolation of resource counts from
+			// apply-time-interpolated expressions, such as those in
+			// "provisioner" blocks.
+			//
+			// Here we are just populating the interpolated value in-place
+			// inside this RawConfig object, like we would in
+			// NodeAbstractCountResource.
+			&EvalInterpolate{Config: n.Config.RawCount},
+
 			&EvalInterpolate{
 				Config:   n.Config.RawConfig.Copy(),
 				Resource: resource,
diff --git a/terraform/test-fixtures/apply-provisioner-interp-count/provisioner-interp-count.tf b/terraform/test-fixtures/apply-provisioner-interp-count/provisioner-interp-count.tf
new file mode 100644
index 000000000000..f6457a3dd30e
--- /dev/null
+++ b/terraform/test-fixtures/apply-provisioner-interp-count/provisioner-interp-count.tf
@@ -0,0 +1,17 @@
+variable "count" {
+  default = 3
+}
+
+resource "aws_instance" "a" {
+  count = "${var.count}"
+}
+
+resource "aws_instance" "b" {
+  provisioner "local-exec" {
+    # Since we're in a provisioner block here, this interpolation is
+    # resolved during the apply walk and so the resource count must
+    # be interpolated during that walk, even though apply walk doesn't
+    # do DynamicExpand.
+    command = "echo ${aws_instance.a.count}"
+  }
+}