[Python] AWS MSK Firehose to S3 data loss

I have an AWS MSK Kafka serverless cluster with an AWS Firehose sinking messages directly to S3, the entire set up is done in AWS private VPC.

My Kafka producer code is as below

from kafka import KafkaProducer
from aws_msk_iam_sasl_signer import MSKAuthTokenProvider
import json
import csv


class MSKTokenProvider():
def token(self):
token, _ = MSKAuthTokenProvider.generate_auth_token(REGION)
return token

tp = MSKTokenProvider()

producer = KafkaProducer(
bootstrap_servers=BROKERS,
value_serializer=lambda v: json.dumps(v).encode('utf-8'),
retry_backoff_ms=500,
request_timeout_ms=20000,
security_protocol='SASL_SSL',
sasl_mechanism='OAUTHBEARER',
sasl_oauth_token_provider=tp,
acks = 'all')

def read_csv_chunk(csv_file_path,batch_size):
try:
with open(csv_file_path, mode='r', encoding='utf-8-sig') as file:
reader = csv.DictReader(file)
batch = []

for i, row in enumerate(reader):
batch.append(row)

# When batch is full, publish to Kafka
if len(batch) >= batch_size:
publish_batch(batch)
batch = [] # Clear batch buffer

# Log progress every 10,000 rows
if i % batch_size == 0:
print(f"{i} rows processed...")

# Publish remaining rows in batch
if batch:
publish_batch(batch)

except Exception as e:
print(f"Error processing file: {e}")

finally:
# Flush all pending messages
producer.flush()

def publish_batch(batch):
"""Publish a batch of messages to Kafka."""
for row in batch:
try:
producer.send(TOPIC_NAME, value=row)
except Exception as e:
print(f"Error sending message: {e}")

if __name__ == '__main__':

csv_file_path = 'data.csv' # Path to CSV file
batch_size = 10000 # message batch size

read_csv_chunk(csv_file_path,batch_size) # Process the CSV file in batch and publish to Kafka


I am sending 2.6 million data over to the topic, and no filtering is applied. By right, I should receive 2.6 million data in S3, but I noticed that if I don't add the last acks = 'all', then I would only receive around 1.9 million data in S3, only when acks = 'all' can I receive the complete copy.

From what I understand, when I put acks = 'all' , all brokers MUST ack the message, otherwise Kakfa will redeliver the message, that's why I can receive the whole copy.

My question is: when I don't have acks = 'all', acks should default to 1, and my code runs without any error. Does that mean all messages have been delivered successfully? Or at least one broker has received the message, which means I should not have any data loss, but the actual case is the opposite.

Or is it because serverless cluster only maintain 1 broker, so when the message is not deliver, then it is not delivered, there is no second broker for back up?

I really appreciate any thoughts or corrections.

Continue reading...