Fix Kafka input for filebeat (#30277)

The Kafka input was broken and had 2 issues:

A serialization error on filebeat output:

Looks like the ack function was put in the `reader.Message.Meta` map by mistake in
20d6038
`Meta` is a `MapStr` type that does not support function values, therefore fails to
serialize itself when requested later on output.

`ack` was not called for incoming messages:

The ack function was never used because it was supposed
to be a part of the `beat.Event.Private` and it was not put in there.

(cherry picked from commit c2f51de)

CHANGELOG.next.asciidoc

@@ -39,6 +39,17 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 
 *Auditbeat*
 
+- auditd: Add error.message to events when processing fails. {pull}30009[30009]
+
+*Filebeat*
+
+- tcp/unix input: Stop accepting connections after socket is closed. {pull}29712[29712]
+- Fix using log_group_name_prefix in aws-cloudwatch input. {pull}29695[29695]
+- aws-s3: fix race condition in states used by s3-poller. {issue}30123[30123] {pull}30131[30131]
+- Fix broken Kafka input {issue}29746[29746] {pull}30277[30277]
+
+*Heartbeat*
+
 
 *Filebeat*
 

filebeat/input/kafka/input.go

@@ -318,6 +318,7 @@ func (h *groupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim s
 			Timestamp: message.Ts,
 			Meta:      message.Meta,
 			Fields:    message.Fields,
+			Private:   message.Private,
 		})
 	}
 	return nil
@@ -458,8 +459,8 @@ func composeMessage(timestamp time.Time, content []byte, kafkaFields common.MapS
 			"kafka":   kafkaFields,
 			"message": string(content),
 		},
-		Meta: common.MapStr{
-			"ackHandler": ackHandler,
+		Private: eventMeta{
+			ackHandler: ackHandler,
 		},
 	}
 }

filebeat/input/kafka/kafka_integration_test.go

@@ -62,6 +62,7 @@ func recordHeader(key, value string) sarama.RecordHeader {
 
 func TestInput(t *testing.T) {
 	testTopic := createTestTopicName()
+	groupID := "filebeat"
 
 	// Send test messages to the topic for the input to read.
 	messages := []testMessage{
@@ -88,7 +89,7 @@ func TestInput(t *testing.T) {
 	config := common.MustNewConfigFrom(common.MapStr{
 		"hosts":      getTestKafkaHost(),
 		"topics":     []string{testTopic},
-		"group_id":   "filebeat",
+		"group_id":   groupID,
 		"wait_close": 0,
 	})
 
@@ -113,6 +114,13 @@ func TestInput(t *testing.T) {
 			assert.Equal(t, text, msg.message)
 
 			checkMatchingHeaders(t, event, msg.headers)
+
+			// emulating the pipeline (kafkaInput.Run)
+			meta, ok := event.Private.(eventMeta)
+			if !ok {
+				t.Fatal("could not get eventMeta and ack the message")
+			}
+			meta.ackHandler()
 		case <-timeout:
 			t.Fatal("timeout waiting for incoming events")
 		}
@@ -132,6 +140,8 @@ func TestInput(t *testing.T) {
 		t.Fatal("timeout waiting for beat to shut down")
 	case <-didClose:
 	}
+
+	assertOffset(t, groupID, testTopic, int64(len(messages)))
 }
 
 func TestInputWithMultipleEvents(t *testing.T) {
@@ -420,6 +430,33 @@ func getTestKafkaHost() string {
 	)
 }
 
+func assertOffset(t *testing.T, groupID, topic string, expected int64) {
+	client, err := sarama.NewClient([]string{getTestKafkaHost()}, nil)
+	assert.NoError(t, err)
+	defer client.Close()
+
+	ofm, err := sarama.NewOffsetManagerFromClient(groupID, client)
+	assert.NoError(t, err)
+	defer ofm.Close()
+
+	partitions, err := client.Partitions(topic)
+	assert.NoError(t, err)
+
+	var offsetSum int64
+
+	for _, partitionID := range partitions {
+		pom, err := ofm.ManagePartition(topic, partitionID)
+		assert.NoError(t, err)
+
+		offset, _ := pom.NextOffset()
+		offsetSum += offset
+
+		pom.Close()
+	}
+
+	assert.Equal(t, expected, offsetSum, "offset does not match, perhaps messages were not acknowledged")
+}
+
 func writeToKafkaTopic(
 	t *testing.T, topic string, message string,
 	headers []sarama.RecordHeader, timeout time.Duration,