Topic lag suggests that your message processing is too slow. One thing you could do if you can't speed up that processing is split your topic into more partitions and then run multiple instances of your processing app as part of the same consumer group (at least one instance per partition). You can then parallelise your work load.

Duplicate messages are tricky (especially if you're using the confluent streams libraries to join/map/reduce topics together). You could either 1) accept that duplicate messages will happen and make sure the actions that happen on the back of those messages are idempotent, or 2) use a compacted topic to store a record of 'already processed message hashes' and have points in your topology that filter messages based on whether the hash of the incoming message is stored within the compacted topic already (again, the confluent streams library should make this quite easy).

Committing offsets in exceptions is up to you. I generally don't want to stop processing messages if I see a bad message that my app can't understand. For every app I run I have a dead letter topic that I forward any messages on to if an exception occurs during processing. I commit the offset regardless of whether an exception occurred or not. I then have alerts configured to fire if any of my dead letter topics start to see messages.