这个问题的出现是因为Lambda函数没有正确处理Kafka的消息。其结果是函数不断地尝试重新处理相同的消息，导致无限循环。

以下是一种可能的解决方法，通过使用“记录”功能，将Lambda函数的状态存储在DynamoDB表中，这样每条消息只会被消费一次：

import boto3
import json
import logging
import time

logger = logging.getLogger()
logger.setLevel(logging.INFO)
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('lambda_state')

def lambda_handler(event, context):
    logger.info('Received event: ' + json.dumps(event))

    # Check if message has already been processed
    msg_id = event['Records'][0]['SNS']['MessageId']
    response = table.get_item(Key={'id': msg_id})

    if 'Item' not in response:
        # Process message
        logger.info('Processing message: ' + json.dumps(event))

        # Save state
        table.put_item(Item={'id': msg_id, 'timestamp': int(time.time())})

    else:
        logger.info('Message already processed: ' + json.dumps(event))

    return

此解决方案使用DynamoDB表来存储已处理的消息列表，并检查新的Kafka消息是否在此列表中。如果不在，则处理消息，并将其添加到DynamoDB表中。如果在，则跳过处理。

请注意，此方法仅使用了一个DynamoDB表。但是，在实际生产中，您可能需要使用多个表，具体取决于应用程序的要求和设计。