Add config check to replace processor

CHANGELOG.next.asciidoc

@@ -96,6 +96,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Fix cache processor expiries infinite growth when large a large TTL is used and recurring keys are cached. {pull}38561[38561]
 - Fix parsing of RFC 3164 process IDs in syslog processor. {issue}38947[38947] {pull}38982[38982]
 - Rename the field "apache2.module.error" to "apache.module.error" in Apache error visualization. {issue}39480[39480] {pull}39481[39481]
+- Validate config of the `replace` processor {pull}40047[40047]
 
 *Auditbeat*
 

libbeat/processors/actions/replace.go

@@ -43,9 +43,9 @@ type replaceStringConfig struct {
 }
 
 type replaceConfig struct {
-	Field       string         `config:"field"`
-	Pattern     *regexp.Regexp `config:"pattern"`
-	Replacement string         `config:"replacement"`
+	Field       string         `config:"field" validate:"required"`
+	Pattern     *regexp.Regexp `config:"pattern" validate:"required"`
+	Replacement string         `config:"replacement" validate:"required"`
 }
 
 func init() {

libbeat/processors/actions/replace_test.go

@@ -22,14 +22,70 @@ import (
 	"regexp"
 	"testing"
 
+	conf "github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/elastic-agent-libs/mapstr"
 )
 
+func TestBadConfig(t *testing.T) {
+	var cases = []struct {
+		name        string
+		cfg         replaceStringConfig
+		shouldError bool
+	}{
+		{
+			name:        "field-only",
+			cfg:         replaceStringConfig{Fields: []replaceConfig{{Field: "message"}}},
+			shouldError: true,
+		},
+		{
+			name:        "no-regex",
+			cfg:         replaceStringConfig{Fields: []replaceConfig{{Field: "message", Replacement: "new_message"}}},
+			shouldError: true,
+		},
+		{
+			name:        "no-replacement",
+			cfg:         replaceStringConfig{Fields: []replaceConfig{{Field: "message", Pattern: regexp.MustCompile(`message`)}}},
+			shouldError: true,
+		},
+		{
+			name: "valid-then-invalid",
+			cfg: replaceStringConfig{Fields: []replaceConfig{
+				{Field: "message", Pattern: regexp.MustCompile(`message`), Replacement: "new_message"},
+				{Field: "message", Pattern: regexp.MustCompile(`message`)},
+			},
+			},
+			shouldError: true,
+		},
+		{
+			name:        "no-error",
+			cfg:         replaceStringConfig{Fields: []replaceConfig{{Field: "message", Replacement: "new_message", Pattern: regexp.MustCompile(`message`)}}},
+			shouldError: false,
+		},
+	}
+
+	for _, testCase := range cases {
+		t.Run(testCase.name, func(t *testing.T) {
+			cfg, err := conf.NewConfigFrom(testCase.cfg)
+			require.NoError(t, err)
+			unpacked := replaceStringConfig{}
+			err = cfg.Unpack(&unpacked)
+			if testCase.shouldError {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+
+		})
+	}
+
+}
+
 func TestReplaceRun(t *testing.T) {
 	var tests = []struct {
 		description   string