Add producer auto resolution case

[vahidhashemian]

We've observed some producer applications hitting this error during broker replacements that is a sign of them losing the up to date partition leadership information:

Failed to send data via PSC producer: Expiring xx record(s) for <topic>-<partition>:320000 ms has passed since batch creation
	at com.pinterest.flink.streaming.connectors.psc.FlinkPscProducer.checkErroneous(FlinkPscProducer.java:1847)
	at com.pinterest.flink.streaming.connectors.psc.FlinkPscProducer.invoke(FlinkPscProducer.java:856)
	at com.pinterest.flink.streaming.connectors.psc.FlinkPscProducer.invoke(FlinkPscProducer.java:111)
	at org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction.invoke(TwoPhaseCommitSinkFunction.java:245)
	at org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:54)
	at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:233)
	at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:134)
	at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:105)
	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:519)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:203)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:807)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:756)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:948)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:927)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:741)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:563)
	at java.lang.Thread.run(Thread.java:748)
...

Caused by: org.apache.kafka.common.errors.TimeoutException: Expiring xx record(s) for -:320000 ms has passed since batch creation