I am not familiar with Apache-Beam but I believe the issue might be related to Kafka not committing the offsets in time. After writing data to the database, the offsets may not be committed immediately. Kafka waits until all messages in the batch (by default 500 messages due to the max.poll.records setting) are processed. This could lead to reprocessing if the system fails or restarts before the offsets are committed.

To reduce the number of duplicates processed in case of failure, you can:

1. Reduce max.poll.records: This will limit the number of messages in each batch, meaning that fewer messages will be at risk of duplication if something goes wrong mid-batch.
2. Manually commit offsets: After each database write, commit the Kafka offsets manually to ensure messages are not reprocessed after they are rebalanced.

For more information, you can refer to this article which explains it nicely Offset Management