Support data replication

### Motivation

For some jobs, data loss and reproduction is not acceptable, data replication is needed to handle data loss in this scenario. 

### Changes

Allow to config the replication factor and only produce the data when all replicas are lost.

### Test
- Unit test.
- Test manually on a cluster.