diff --git a/.changelog/4873.txt b/.changelog/4873.txt
new file mode 100644
index 0000000000..6cd559f9d3
--- /dev/null
+++ b/.changelog/4873.txt
@@ -0,0 +1,3 @@
+```release-note:enhancement
+composer: added field `max_pods_per_node` to resource `google_composer_environment` (beta)
+```
diff --git a/google-beta/resource_composer_environment.go b/google-beta/resource_composer_environment.go
index 5ab3dd7978..aa970e00e9 100644
--- a/google-beta/resource_composer_environment.go
+++ b/google-beta/resource_composer_environment.go
@@ -192,6 +192,14 @@ func resourceComposerEnvironment() *schema.Resource {
 										DiffSuppressFunc: compareServiceAccountEmailToLink,
 										Description:      `The Google Cloud Platform Service Account to be used by the node VMs. If a service account is not specified, the "default" Compute Engine service account is used. Cannot be updated. If given, note that the service account must have roles/composer.worker for any GCP resources created under the Cloud Composer Environment.`,
 									},
+
+									"max_pods_per_node": {
+										Type:        schema.TypeInt,
+										Computed:    true,
+										Optional:    true,
+										ForceNew:    true,
+										Description: `The maximum pods per node in the GKE cluster allocated during environment creation. Lowering this value reduces IP address consumption by the Cloud Composer Kubernetes cluster. This value can only be set during environment creation, and only if the environment is VPC-Native. The range of possible values is 8-110, and the default is 32.`,
+									},
 									"tags": {
 										Type:     schema.TypeSet,
 										Optional: true,
@@ -913,6 +921,8 @@ func flattenComposerEnvironmentConfigNodeConfig(nodeCfg *composer.NodeConfig) in
 	transformed["disk_size_gb"] = nodeCfg.DiskSizeGb
 	transformed["service_account"] = nodeCfg.ServiceAccount
 	transformed["oauth_scopes"] = flattenComposerEnvironmentConfigNodeConfigOauthScopes(nodeCfg.OauthScopes)
+
+	transformed["max_pods_per_node"] = nodeCfg.MaxPodsPerNode
 	transformed["tags"] = flattenComposerEnvironmentConfigNodeConfigTags(nodeCfg.Tags)
 	transformed["ip_allocation_policy"] = flattenComposerEnvironmentConfigNodeConfigIPAllocationPolicy(nodeCfg.IpAllocationPolicy)
 	return []interface{}{transformed}
@@ -1153,6 +1163,10 @@ func expandComposerEnvironmentConfigNodeConfig(v interface{}, d *schema.Resource
 		transformed.ServiceAccount = transformedServiceAccount
 	}
 
+	if transformedMaxPodsPerNode, ok := original["max_pods_per_node"]; ok {
+		transformed.MaxPodsPerNode = int64(transformedMaxPodsPerNode.(int))
+	}
+
 	var nodeConfigZone string
 	if v, ok := original["zone"]; ok {
 		transformedZone, err := expandComposerEnvironmentZone(v, d, config)
diff --git a/google-beta/resource_composer_environment_test.go b/google-beta/resource_composer_environment_test.go
index a985bc217c..02361a10c4 100644
--- a/google-beta/resource_composer_environment_test.go
+++ b/google-beta/resource_composer_environment_test.go
@@ -902,6 +902,11 @@ resource "google_composer_environment" "test" {
 			zone       = "us-central1-a"
 
 			service_account = google_service_account.test.name
+			max_pods_per_node = 33
+			ip_allocation_policy {
+				use_ip_aliases          = true
+				cluster_ipv4_cidr_block = "10.0.0.0/16"
+			}
 		}
 	}
 
diff --git a/google-beta/resource_gke_hub_feature_membership_test.go b/google-beta/resource_gke_hub_feature_membership_test.go
index 32fa84c196..9e1f2ea64d 100644
--- a/google-beta/resource_gke_hub_feature_membership_test.go
+++ b/google-beta/resource_gke_hub_feature_membership_test.go
@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"testing"
 
-	dcl "github.com/GoogleCloudPlatform/declarative-resource-client-library/dcl"
+	"github.com/GoogleCloudPlatform/declarative-resource-client-library/dcl"
 	gkehub "github.com/GoogleCloudPlatform/declarative-resource-client-library/services/google/gkehub/beta"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/terraform"
diff --git a/website/docs/r/composer_environment.html.markdown b/website/docs/r/composer_environment.html.markdown
index d764870acc..18a4459ae6 100644
--- a/website/docs/r/composer_environment.html.markdown
+++ b/website/docs/r/composer_environment.html.markdown
@@ -247,6 +247,14 @@ The `node_config` block supports:
   Structure is documented below.
   Cannot be updated.
 
+* `max_pods_per_node` -
+  (Optional, [Beta](https://terraform.io/docs/providers/google/guides/provider_versions.html))
+  The maximum pods per node in the GKE cluster allocated during environment 
+  creation. Lowering this value reduces IP address consumption by the Cloud 
+  Composer Kubernetes cluster. This value can only be set if the environment is VPC-Native. 
+  The range of possible values is 8-110, and the default is 32.
+  Cannot be updated.
+
 The `software_config` block supports:
 
 * `airflow_config_overrides` -