diff --git a/CHANGELOG.md b/CHANGELOG.md
index cfca62ab..01e752fa 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,10 @@
 
 ## [MAJOR.MINOR.PATCH] - YYYY-MM-DD
 
+- Add `Kafka` field `userConfig.kafka.transaction_partition_verification_enable`, type `boolean`: Enable
+  verification that checks that the partition has been added to the transaction before writing transactional
+  records to the partition
+
 ## v0.15.0 - 2023-11-17
 
 - Upgrade to Go 1.21
diff --git a/api/v1alpha1/userconfig/service/kafka/kafka.go b/api/v1alpha1/userconfig/service/kafka/kafka.go
index 207205f2..2df79133 100644
--- a/api/v1alpha1/userconfig/service/kafka/kafka.go
+++ b/api/v1alpha1/userconfig/service/kafka/kafka.go
@@ -207,6 +207,9 @@ type Kafka struct {
 	// The maximum number of bytes in a socket request (defaults to 104857600).
 	SocketRequestMaxBytes *int `groups:"create,update" json:"socket_request_max_bytes,omitempty"`
 
+	// Enable verification that checks that the partition has been added to the transaction before writing transactional records to the partition
+	TransactionPartitionVerificationEnable *bool `groups:"create,update" json:"transaction_partition_verification_enable,omitempty"`
+
 	// +kubebuilder:validation:Minimum=600000
 	// +kubebuilder:validation:Maximum=3600000
 	// The interval at which to remove transactions that have expired due to transactional.id.expiration.ms passing (defaults to 3600000 (1 hour)).
diff --git a/charts/aiven-operator-crds/templates/aiven.io_kafkas.yaml b/charts/aiven-operator-crds/templates/aiven.io_kafkas.yaml
index 6aa902af..91fb60db 100644
--- a/charts/aiven-operator-crds/templates/aiven.io_kafkas.yaml
+++ b/charts/aiven-operator-crds/templates/aiven.io_kafkas.yaml
@@ -503,6 +503,11 @@ spec:
                         maximum: 209715200
                         minimum: 10485760
                         type: integer
+                      transaction_partition_verification_enable:
+                        description: Enable verification that checks that the partition
+                          has been added to the transaction before writing transactional
+                          records to the partition
+                        type: boolean
                       transaction_remove_expired_transaction_cleanup_interval_ms:
                         description: The interval at which to remove transactions
                           that have expired due to transactional.id.expiration.ms
diff --git a/config/crd/bases/aiven.io_kafkas.yaml b/config/crd/bases/aiven.io_kafkas.yaml
index 6aa902af..91fb60db 100644
--- a/config/crd/bases/aiven.io_kafkas.yaml
+++ b/config/crd/bases/aiven.io_kafkas.yaml
@@ -503,6 +503,11 @@ spec:
                         maximum: 209715200
                         minimum: 10485760
                         type: integer
+                      transaction_partition_verification_enable:
+                        description: Enable verification that checks that the partition
+                          has been added to the transaction before writing transactional
+                          records to the partition
+                        type: boolean
                       transaction_remove_expired_transaction_cleanup_interval_ms:
                         description: The interval at which to remove transactions
                           that have expired due to transactional.id.expiration.ms
diff --git a/docs/docs/api-reference/kafka.md b/docs/docs/api-reference/kafka.md
index 7addf479..17de83a8 100644
--- a/docs/docs/api-reference/kafka.md
+++ b/docs/docs/api-reference/kafka.md
@@ -213,6 +213,7 @@ Kafka broker configuration values.
 - [`sasl_oauthbearer_jwks_endpoint_url`](#spec.userConfig.kafka.sasl_oauthbearer_jwks_endpoint_url-property){: name='spec.userConfig.kafka.sasl_oauthbearer_jwks_endpoint_url-property'} (string, MaxLength: 2048). OIDC JWKS endpoint URL. By setting this the SASL SSL OAuth2/OIDC authentication is enabled. See also other options for SASL OAuth2/OIDC.
 - [`sasl_oauthbearer_sub_claim_name`](#spec.userConfig.kafka.sasl_oauthbearer_sub_claim_name-property){: name='spec.userConfig.kafka.sasl_oauthbearer_sub_claim_name-property'} (string, MaxLength: 128). Name of the scope from which to extract the subject claim from the JWT. Defaults to sub.
 - [`socket_request_max_bytes`](#spec.userConfig.kafka.socket_request_max_bytes-property){: name='spec.userConfig.kafka.socket_request_max_bytes-property'} (integer, Minimum: 10485760, Maximum: 209715200). The maximum number of bytes in a socket request (defaults to 104857600).
+- [`transaction_partition_verification_enable`](#spec.userConfig.kafka.transaction_partition_verification_enable-property){: name='spec.userConfig.kafka.transaction_partition_verification_enable-property'} (boolean). Enable verification that checks that the partition has been added to the transaction before writing transactional records to the partition.
 - [`transaction_remove_expired_transaction_cleanup_interval_ms`](#spec.userConfig.kafka.transaction_remove_expired_transaction_cleanup_interval_ms-property){: name='spec.userConfig.kafka.transaction_remove_expired_transaction_cleanup_interval_ms-property'} (integer, Minimum: 600000, Maximum: 3600000). The interval at which to remove transactions that have expired due to transactional.id.expiration.ms passing (defaults to 3600000 (1 hour)).
 - [`transaction_state_log_segment_bytes`](#spec.userConfig.kafka.transaction_state_log_segment_bytes-property){: name='spec.userConfig.kafka.transaction_state_log_segment_bytes-property'} (integer, Minimum: 1048576, Maximum: 2147483647). The transaction topic segment bytes should be kept relatively small in order to facilitate faster log compaction and cache loads (defaults to 104857600 (100 mebibytes)).