diff --git a/changelogs/8.15.asciidoc b/changelogs/8.15.asciidoc
index cefc7469a98..1b205224465 100644
--- a/changelogs/8.15.asciidoc
+++ b/changelogs/8.15.asciidoc
@@ -35,3 +35,4 @@ https://github.com/elastic/apm-server/compare/v8.14.3\...v8.15.0[View commits]
 - Add require data stream to bulk index requests {pull}13398[13398]
 - Support self-instrumentation when in managed mode by getting tracing configs via reloader {pull}13514[13514] {pull}13653[13653]
 - Add mapping for OpenTelemetry attribute `messaging.destination.name` to derive `service.target` correctly {pull}13472[13472]
+- APM Server now automatically retries document-level 429s from Elasticsearch to avoid dropping data. `output.elasticsearch.max_retries` now controls both request-level and document-level retries, and defaults to `3`. {pull}13620[13620]
diff --git a/internal/beater/beater.go b/internal/beater/beater.go
index da7327920a5..3c94c336086 100644
--- a/internal/beater/beater.go
+++ b/internal/beater/beater.go
@@ -774,7 +774,7 @@ func (s *Runner) newFinalBatchProcessor(
 func (s *Runner) newDocappenderConfig(tracer *apm.Tracer, memLimit float64) (
 	docappender.Config, *elasticsearch.Config, error,
 ) {
-	var esConfig struct {
+	esConfig := struct {
 		*elasticsearch.Config `config:",inline"`
 		FlushBytes            string        `config:"flush_bytes"`
 		FlushInterval         time.Duration `config:"flush_interval"`
@@ -782,11 +782,12 @@ func (s *Runner) newDocappenderConfig(tracer *apm.Tracer, memLimit float64) (
 		Scaling               struct {
 			Enabled *bool `config:"enabled"`
 		} `config:"autoscaling"`
+	}{
+		// Default to 1mib flushes, which is the default for go-docappender.
+		FlushBytes:    "1 mib",
+		FlushInterval: time.Second,
+		Config:        elasticsearch.DefaultConfig(),
 	}
-	// Default to 1mib flushes, which is the default for go-docappender.
-	esConfig.FlushBytes = "1 mib"
-	esConfig.FlushInterval = time.Second
-	esConfig.Config = elasticsearch.DefaultConfig()
 	esConfig.MaxIdleConnsPerHost = 10
 
 	if err := s.elasticsearchOutputConfig.Unpack(&esConfig); err != nil {
@@ -819,6 +820,10 @@ func (s *Runner) newDocappenderConfig(tracer *apm.Tracer, memLimit float64) (
 		Scaling:           scalingCfg,
 		Logger:            zap.New(s.logger.Core(), zap.WithCaller(true)),
 		RequireDataStream: true,
+		// Use the output's max_retries to configure the go-docappender's
+		// document level retries.
+		MaxDocumentRetries:    esConfig.MaxRetries,
+		RetryOnDocumentStatus: []int{429}, // Only retry "safe" 429 responses.
 	}, memLimit, s.logger)
 	if cfg.MaxRequests != 0 {
 		esConfig.MaxIdleConnsPerHost = cfg.MaxRequests
diff --git a/internal/beater/beater_test.go b/internal/beater/beater_test.go
index b510a89b066..c7aa453c163 100644
--- a/internal/beater/beater_test.go
+++ b/internal/beater/beater_test.go
@@ -181,13 +181,15 @@ func TestRunnerNewDocappenderConfig(t *testing.T) {
 			docCfg, esCfg, err := r.newDocappenderConfig(nil, c.memSize)
 			require.NoError(t, err)
 			assert.Equal(t, docappender.Config{
-				Logger:             zap.New(r.logger.Core(), zap.WithCaller(true)),
-				CompressionLevel:   5,
-				RequireDataStream:  true,
-				FlushInterval:      time.Second,
-				FlushBytes:         1024 * 1024,
-				MaxRequests:        c.wantMaxRequests,
-				DocumentBufferSize: c.wantDocBufSize,
+				Logger:                zap.New(r.logger.Core(), zap.WithCaller(true)),
+				CompressionLevel:      5,
+				RequireDataStream:     true,
+				FlushInterval:         time.Second,
+				FlushBytes:            1024 * 1024,
+				MaxRequests:           c.wantMaxRequests,
+				DocumentBufferSize:    c.wantDocBufSize,
+				MaxDocumentRetries:    3,
+				RetryOnDocumentStatus: []int{429},
 			}, docCfg)
 			assert.Equal(t, &elasticsearch.Config{
 				Hosts:               elasticsearch.Hosts{"localhost:9200"},
@@ -211,13 +213,15 @@ func TestRunnerNewDocappenderConfig(t *testing.T) {
 			docCfg, esCfg, err := r.newDocappenderConfig(nil, c.memSize)
 			require.NoError(t, err)
 			assert.Equal(t, docappender.Config{
-				Logger:             zap.New(r.logger.Core(), zap.WithCaller(true)),
-				CompressionLevel:   5,
-				RequireDataStream:  true,
-				FlushInterval:      2 * time.Second,
-				FlushBytes:         500 * 1024,
-				MaxRequests:        50,
-				DocumentBufferSize: c.wantDocBufSize,
+				Logger:                zap.New(r.logger.Core(), zap.WithCaller(true)),
+				CompressionLevel:      5,
+				RequireDataStream:     true,
+				FlushInterval:         2 * time.Second,
+				FlushBytes:            500 * 1024,
+				MaxRequests:           50,
+				DocumentBufferSize:    c.wantDocBufSize,
+				MaxDocumentRetries:    3,
+				RetryOnDocumentStatus: []int{429},
 			}, docCfg)
 			assert.Equal(t, &elasticsearch.Config{
 				Hosts:               elasticsearch.Hosts{"localhost:9200"},