diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
index 1c73e56a5ec..d6cb347f636 100644
--- a/CHANGELOG.next.asciidoc
+++ b/CHANGELOG.next.asciidoc
@@ -138,6 +138,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add FIPS configuration option for all AWS API calls. {pull}28899[28899]
 - Add `default_region` config to AWS common module. {pull}29415[29415]
 - Add support for latest k8s versions v1.23 and v1.22 {pull}29575[29575]
+- Only connect to Elasticsearch instances with the same version or newer. {pull}29683[29683]
 
 *Auditbeat*
 
diff --git a/auditbeat/auditbeat.reference.yml b/auditbeat/auditbeat.reference.yml
index 27be08dd194..cce2bc58144 100644
--- a/auditbeat/auditbeat.reference.yml
+++ b/auditbeat/auditbeat.reference.yml
@@ -460,6 +460,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # auditbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 
diff --git a/filebeat/Jenkinsfile.yml b/filebeat/Jenkinsfile.yml
index f5843c43b18..b77dc29f160 100644
--- a/filebeat/Jenkinsfile.yml
+++ b/filebeat/Jenkinsfile.yml
@@ -48,6 +48,7 @@ stages:
       mage: >-                 ## Run module integration tests under previous minor of ES to ensure ingest pipeline compatibility.
         STACK_ENVIRONMENT=prev-minor
         TESTING_FILEBEAT_SKIP_DIFF=1
+        TESTING_FILEBEAT_ALLOW_OLDER=1
         PYTEST_ADDOPTS='-k test_modules'
         mage pythonIntegTest
       withModule: true
diff --git a/filebeat/filebeat.reference.yml b/filebeat/filebeat.reference.yml
index 87a50f91a08..11273253e6a 100644
--- a/filebeat/filebeat.reference.yml
+++ b/filebeat/filebeat.reference.yml
@@ -1404,6 +1404,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # filebeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 
diff --git a/filebeat/tests/system/test_modules.py b/filebeat/tests/system/test_modules.py
index b4ee0c7b487..213a5c04024 100644
--- a/filebeat/tests/system/test_modules.py
+++ b/filebeat/tests/system/test_modules.py
@@ -124,6 +124,9 @@ def run_on_file(self, module, fileset, test_file, cfgfile):
                 module=module, fileset=fileset, test_file=test_file),
             "-M", "*.*.input.close_eof=true",
         ]
+        # allow connecting older versions of Elasticsearch
+        if os.getenv("TESTING_FILEBEAT_ALLOW_OLDER"):
+            cmd.extend(["-E", "output.elasticsearch.allow_older_versions=true"])
 
         # Based on the convention that if a name contains -json the json format is needed. Currently used for LS.
         if "-json" in test_file:
diff --git a/heartbeat/heartbeat.reference.yml b/heartbeat/heartbeat.reference.yml
index aca74cd9e63..415ca4cb69d 100644
--- a/heartbeat/heartbeat.reference.yml
+++ b/heartbeat/heartbeat.reference.yml
@@ -606,6 +606,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # heartbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 
diff --git a/libbeat/_meta/config/output-elasticsearch.reference.yml.tmpl b/libbeat/_meta/config/output-elasticsearch.reference.yml.tmpl
index 943e6a16b47..be9ede7dba8 100644
--- a/libbeat/_meta/config/output-elasticsearch.reference.yml.tmpl
+++ b/libbeat/_meta/config/output-elasticsearch.reference.yml.tmpl
@@ -77,6 +77,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # {{.BeatName}} expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
 {{include "ssl.reference.yml.tmpl" . | indent 2 }}
   # Enable Kerberos support. Kerberos is automatically enabled if any Kerberos setting is set.
   #kerberos.enabled: true
diff --git a/libbeat/cmd/instance/beat.go b/libbeat/cmd/instance/beat.go
index eb126966094..ab52b4f545c 100644
--- a/libbeat/cmd/instance/beat.go
+++ b/libbeat/cmd/instance/beat.go
@@ -326,6 +326,8 @@ func (b *Beat) createBeater(bt beat.Creator) (beat.Beater, error) {
 	logSystemInfo(b.Info)
 	logp.Info("Setup Beat: %s; Version: %s", b.Info.Beat, b.Info.Version)
 
+	b.checkElasticsearchVersion()
+
 	err = b.registerESIndexManagement()
 	if err != nil {
 		return nil, err
@@ -855,6 +857,37 @@ func (b *Beat) loadDashboards(ctx context.Context, force bool) error {
 	return nil
 }
 
+// checkElasticsearchVersion registers a global callback to make sure ES instance we are connecting
+// to is at least on the same version as the Beat.
+// If the check is disabled or the output is not Elasticsearch, nothing happens.
+func (b *Beat) checkElasticsearchVersion() {
+	if b.Config.Output.Name() != "elasticsearch" || b.isConnectionToOlderVersionAllowed() {
+		return
+	}
+
+	elasticsearch.RegisterGlobalCallback(func(conn *eslegclient.Connection) error {
+		esVersion := conn.GetVersion()
+		beatVersion, err := common.NewVersion(b.Info.Version)
+		if err != nil {
+			return err
+		}
+		if esVersion.LessThan(beatVersion) {
+			return fmt.Errorf("%v ES=%s, Beat=%s.", elasticsearch.ErrTooOld, esVersion.String(), b.Info.Version)
+		}
+		return nil
+	})
+}
+
+func (b *Beat) isConnectionToOlderVersionAllowed() bool {
+	config := struct {
+		AllowOlder bool `config:"allow_older_versions"`
+	}{false}
+
+	b.Config.Output.Config().Unpack(&config)
+
+	return config.AllowOlder
+}
+
 // registerESIndexManagement registers the loading of the template and ILM
 // policy as a callback with the elasticsearch output. It is important the
 // registration happens before the publisher is created.
diff --git a/libbeat/outputs/elasticsearch/client.go b/libbeat/outputs/elasticsearch/client.go
index 0d9d619b9f5..f13a5db56ec 100644
--- a/libbeat/outputs/elasticsearch/client.go
+++ b/libbeat/outputs/elasticsearch/client.go
@@ -38,7 +38,11 @@ import (
 	"github.com/elastic/beats/v7/libbeat/testing"
 )
 
-var errPayloadTooLarge = errors.New("the bulk payload is too large for the server. Consider to adjust `http.max_content_length` parameter in Elasticsearch or `bulk_max_size` in the beat. The batch has been dropped")
+var (
+	errPayloadTooLarge = errors.New("the bulk payload is too large for the server. Consider to adjust `http.max_content_length` parameter in Elasticsearch or `bulk_max_size` in the beat. The batch has been dropped")
+
+	ErrTooOld = errors.New("Elasticsearch is too old. Please upgrade the instance. If you would like to connect to older instances set output.elasticsearch.allow_older_versions to true.")
+)
 
 // Client is an elasticsearch client.
 type Client struct {
diff --git a/libbeat/outputs/elasticsearch/config.go b/libbeat/outputs/elasticsearch/config.go
index 816078f9225..4e61467f679 100644
--- a/libbeat/outputs/elasticsearch/config.go
+++ b/libbeat/outputs/elasticsearch/config.go
@@ -43,6 +43,7 @@ type elasticsearchConfig struct {
 	MaxRetries         int                     `config:"max_retries"`
 	Backoff            Backoff                 `config:"backoff"`
 	NonIndexablePolicy *common.ConfigNamespace `config:"non_indexable_policy"`
+	AllowOlderVersion  bool                    `config:"allow_older_versions"`
 
 	Transport httpcommon.HTTPTransportSettings `config:",inline"`
 }
diff --git a/libbeat/outputs/elasticsearch/docs/elasticsearch.asciidoc b/libbeat/outputs/elasticsearch/docs/elasticsearch.asciidoc
index a3c89dad67e..c465be2dddd 100644
--- a/libbeat/outputs/elasticsearch/docs/elasticsearch.asciidoc
+++ b/libbeat/outputs/elasticsearch/docs/elasticsearch.asciidoc
@@ -60,6 +60,9 @@ This output works with all compatible versions of Elasticsearch. See the
 https://www.elastic.co/support/matrix#matrix_compatibility[Elastic Support
 Matrix].
 
+For optimal experience, {beatname_uc} only connects to instances that are at least on the
+same version as the Beat. The check can be disabled by setting `output.elasticsearch.allow_older_versions`.
+
 ==== Configuration options
 
 You can specify the following options in the `elasticsearch` section of the +{beatname_lc}.yml+ config file:
@@ -668,6 +671,14 @@ Elasticsearch after a network error. The default is `60s`.
 
 The http request timeout in seconds for the Elasticsearch request. The default is 90.
 
+==== `allow_older_versions`
+
+By default, {beatname_uc} expects the Elasticsearch instance to be on the same or newer version to provide
+optimal experience. We suggest you connect to the same version to make sure all features {beatname_uc} is using are
+available in your Elasticsearch instance.
+
+You can disable the check for example during updating the Elastic Stack, so data collection can go on.
+
 ===== `ssl`
 
 Configuration options for SSL parameters like the certificate authority to use
diff --git a/libbeat/tests/system/config/mockbeat.yml.j2 b/libbeat/tests/system/config/mockbeat.yml.j2
index 8aee28a276f..5657105832e 100644
--- a/libbeat/tests/system/config/mockbeat.yml.j2
+++ b/libbeat/tests/system/config/mockbeat.yml.j2
@@ -42,6 +42,8 @@ output:
     {% for k, v in elasticsearch.items() -%}
     {{ k }}: {{ v }}
     {% endfor -%}
+    # older versions have to be allowed because mockbeat is on v9.9.9
+    allow_older_versions: true
   {%- endif %}
 
   # Redis as output
diff --git a/metricbeat/metricbeat.reference.yml b/metricbeat/metricbeat.reference.yml
index 2d4c51aea50..b98461466cf 100644
--- a/metricbeat/metricbeat.reference.yml
+++ b/metricbeat/metricbeat.reference.yml
@@ -1323,6 +1323,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # metricbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 
diff --git a/packetbeat/packetbeat.reference.yml b/packetbeat/packetbeat.reference.yml
index 739fe5c5332..fd9ed4b20fe 100644
--- a/packetbeat/packetbeat.reference.yml
+++ b/packetbeat/packetbeat.reference.yml
@@ -955,6 +955,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # packetbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 
diff --git a/winlogbeat/winlogbeat.reference.yml b/winlogbeat/winlogbeat.reference.yml
index 51ca51897f8..21358d6e416 100644
--- a/winlogbeat/winlogbeat.reference.yml
+++ b/winlogbeat/winlogbeat.reference.yml
@@ -391,6 +391,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # winlogbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 
diff --git a/x-pack/auditbeat/auditbeat.reference.yml b/x-pack/auditbeat/auditbeat.reference.yml
index b058641150c..1fd3bfcc92c 100644
--- a/x-pack/auditbeat/auditbeat.reference.yml
+++ b/x-pack/auditbeat/auditbeat.reference.yml
@@ -516,6 +516,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # auditbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 
diff --git a/x-pack/filebeat/Jenkinsfile.yml b/x-pack/filebeat/Jenkinsfile.yml
index c5647d9a854..79276187dfe 100644
--- a/x-pack/filebeat/Jenkinsfile.yml
+++ b/x-pack/filebeat/Jenkinsfile.yml
@@ -48,6 +48,7 @@ stages:
       mage: >-                 ## Run module integration tests under previous minor of ES to ensure ingest pipeline compatibility.
         STACK_ENVIRONMENT=prev-minor
         TESTING_FILEBEAT_SKIP_DIFF=1
+        TESTING_FILEBEAT_ALLOW_OLDER=1
         PYTEST_ADDOPTS='-k test_xpack_modules'
         mage pythonIntegTest
       withModule: true
diff --git a/x-pack/filebeat/filebeat.reference.yml b/x-pack/filebeat/filebeat.reference.yml
index cabffe369c9..7aa5350e19f 100644
--- a/x-pack/filebeat/filebeat.reference.yml
+++ b/x-pack/filebeat/filebeat.reference.yml
@@ -3634,6 +3634,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # filebeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 
diff --git a/x-pack/functionbeat/functionbeat.reference.yml b/x-pack/functionbeat/functionbeat.reference.yml
index 090ab1cc877..5f0e7bfd13e 100644
--- a/x-pack/functionbeat/functionbeat.reference.yml
+++ b/x-pack/functionbeat/functionbeat.reference.yml
@@ -638,6 +638,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # functionbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 
diff --git a/x-pack/heartbeat/heartbeat.reference.yml b/x-pack/heartbeat/heartbeat.reference.yml
index aca74cd9e63..415ca4cb69d 100644
--- a/x-pack/heartbeat/heartbeat.reference.yml
+++ b/x-pack/heartbeat/heartbeat.reference.yml
@@ -606,6 +606,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # heartbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 
diff --git a/x-pack/metricbeat/metricbeat.reference.yml b/x-pack/metricbeat/metricbeat.reference.yml
index 2c3a35d801b..18b6d9704a8 100644
--- a/x-pack/metricbeat/metricbeat.reference.yml
+++ b/x-pack/metricbeat/metricbeat.reference.yml
@@ -1844,6 +1844,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # metricbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 
diff --git a/x-pack/osquerybeat/osquerybeat.reference.yml b/x-pack/osquerybeat/osquerybeat.reference.yml
index 53f7f52f922..b81f31c7d16 100644
--- a/x-pack/osquerybeat/osquerybeat.reference.yml
+++ b/x-pack/osquerybeat/osquerybeat.reference.yml
@@ -357,6 +357,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # osquerybeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 
diff --git a/x-pack/packetbeat/packetbeat.reference.yml b/x-pack/packetbeat/packetbeat.reference.yml
index 739fe5c5332..fd9ed4b20fe 100644
--- a/x-pack/packetbeat/packetbeat.reference.yml
+++ b/x-pack/packetbeat/packetbeat.reference.yml
@@ -955,6 +955,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # packetbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true
 
diff --git a/x-pack/winlogbeat/winlogbeat.reference.yml b/x-pack/winlogbeat/winlogbeat.reference.yml
index d74e971638a..d44f301dd2d 100644
--- a/x-pack/winlogbeat/winlogbeat.reference.yml
+++ b/x-pack/winlogbeat/winlogbeat.reference.yml
@@ -434,6 +434,10 @@ output.elasticsearch:
   # Configure HTTP request timeout before failing a request to Elasticsearch.
   #timeout: 90
 
+  # winlogbeat expects Elasticsearch to be the same version or newer than the Beat.
+  # Lift the version restriction by setting allow_older_versions to true.
+  #allow_older_versions: false
+
   # Use SSL settings for HTTPS.
   #ssl.enabled: true