Kafka: Add support to configure heartbeat interval and session timeout to kafka's consumer

bindings/kafka/metadata.yaml

@@ -270,6 +270,18 @@ metadata:
       Disables consumer retry by setting this to "false".
     example: '"true"'
     default: '"false"'
+  - name: heartbeatInterval
+    type: duration
+    description: |
+      The interval between heartbeats to the consumer coordinator.
+    example: '"5s"'
+    default: '"3s"'
+  - name: sessionTimeout
+    type: duration
+    description: |
+      The maximum time between heartbeats before the consumer is considered inactive and will timeout.
+    example: '"20s"'
+    default: '"10s"'
   - name: version
     type: string
     description: |

common/component/kafka/kafka.go

@@ -141,6 +141,8 @@ func (k *Kafka) Init(ctx context.Context, metadata map[string]string) error {
 	config.Consumer.Offsets.Initial = k.initialOffset
 	config.Consumer.Fetch.Min = meta.consumerFetchMin
 	config.Consumer.Fetch.Default = meta.consumerFetchDefault
+	config.Consumer.Group.Heartbeat.Interval = meta.HeartbeatInterval
+	config.Consumer.Group.Session.Timeout = meta.SessionTimeout
 	config.ChannelBufferSize = meta.channelBufferSize
 
 	config.Net.KeepAlive = meta.ClientConnectionKeepAliveInterval

common/component/kafka/metadata.go

@@ -86,6 +86,8 @@ type KafkaMetadata struct {
 	TLSClientKey           string              `mapstructure:"clientKey"`
 	ConsumeRetryEnabled    bool                `mapstructure:"consumeRetryEnabled"`
 	ConsumeRetryInterval   time.Duration       `mapstructure:"consumeRetryInterval"`
+	HeartbeatInterval      time.Duration       `mapstructure:"heartbeatInterval"`
+	SessionTimeout         time.Duration       `mapstructure:"sessionTimeout"`
 	Version                string              `mapstructure:"version"`
 	internalVersion        sarama.KafkaVersion `mapstructure:"-"`
 	internalOidcExtensions map[string]string   `mapstructure:"-"`
@@ -158,6 +160,8 @@ func (k *Kafka) getKafkaMetadata(meta map[string]string) (*KafkaMetadata, error)
 		consumerFetchDefault:                         1024 * 1024,
 		ClientConnectionTopicMetadataRefreshInterval: defaultClientConnectionTopicMetadataRefreshInterval,
 		ClientConnectionKeepAliveInterval:            defaultClientConnectionKeepAliveInterval,
+		HeartbeatInterval:                            3 * time.Second,
+		SessionTimeout:                               10 * time.Second,
 	}
 
 	err := metadata.DecodeMetadata(meta, &m)

common/component/kafka/metadata_test.go

@@ -61,6 +61,8 @@ func getCompleteMetadata() map[string]string {
 		consumerFetchDefault:   "1048576",
 		consumerFetchMin:       "1",
 		channelBufferSize:      "256",
+		"heartbeatInterval":    "2s",
+		"sessionTimeout":       "30s",
 	}
 }
 
@@ -132,6 +134,8 @@ func assertMetadata(t *testing.T, meta *KafkaMetadata) {
 	require.Equal(t, int32(1024*1024), meta.consumerFetchDefault)
 	require.Equal(t, int32(1), meta.consumerFetchMin)
 	require.Equal(t, 256, meta.channelBufferSize)
+	require.Equal(t, 2*time.Second, meta.HeartbeatInterval)
+	require.Equal(t, 30*time.Second, meta.SessionTimeout)
 	require.Equal(t, 8*time.Minute, defaultClientConnectionTopicMetadataRefreshInterval)
 	require.Equal(t, 0*time.Minute, defaultClientConnectionKeepAliveInterval)
 }
@@ -443,6 +447,64 @@ func TestMetadataChannelBufferSize(t *testing.T) {
 	require.Equal(t, 128, meta.channelBufferSize)
 }
 
+func TestMetadataHeartbeartInterval(t *testing.T) {
+	k := getKafka()
+
+	t.Run("default heartbeat interval", func(t *testing.T) {
+		// arrange
+		m := getBaseMetadata()
+
+		// act
+		meta, err := k.getKafkaMetadata(m)
+
+		// assert
+		require.NoError(t, err)
+		require.Equal(t, 3*time.Second, meta.HeartbeatInterval)
+	})
+
+	t.Run("with heartbeat interval set", func(t *testing.T) {
+		// arrange
+		m := getBaseMetadata()
+		m["heartbeatInterval"] = "1s"
+
+		// act
+		meta, err := k.getKafkaMetadata(m)
+
+		// assert
+		require.NoError(t, err)
+		require.Equal(t, 1*time.Second, meta.HeartbeatInterval)
+	})
+}
+
+func TestMetadataSessionTimeout(t *testing.T) {
+	k := getKafka()
+
+	t.Run("default session timeout", func(t *testing.T) {
+		// arrange
+		m := getBaseMetadata()
+
+		// act
+		meta, err := k.getKafkaMetadata(m)
+
+		// assert
+		require.NoError(t, err)
+		require.Equal(t, 10*time.Second, meta.SessionTimeout)
+	})
+
+	t.Run("with session timeout set", func(t *testing.T) {
+		// arrange
+		m := getBaseMetadata()
+		m["sessionTimeout"] = "20s"
+
+		// act
+		meta, err := k.getKafkaMetadata(m)
+
+		// assert
+		require.NoError(t, err)
+		require.Equal(t, 20*time.Second, meta.SessionTimeout)
+	})
+}
+
 func TestGetEventMetadata(t *testing.T) {
 	ts := time.Now()
 

pubsub/kafka/metadata.yaml

@@ -231,6 +231,18 @@ metadata:
         Disables consumer retry by setting this to "false".
       example: '"true"'
       default: '"false"'
+    - name: heartbeatInterval
+      type: duration
+      description: |
+        The interval between heartbeats to the consumer coordinator.
+      example: '"5s"'
+      default: '"3s"'
+    - name: sessionTimeout
+      type: duration
+      description: |
+        The maximum time between heartbeats before the consumer is considered inactive and will timeout.
+      example: '"20s"'
+      default: '"10s"'
     - name: version
       type: string
       description: |