StreamsProducer should catch InvalidProducerEpoch and throw TaskMigrated in all cases

We fixed the error code handling on producer in https://issues.apache.org/jira/browse/KAFKA-10687, however the newly thrown `InvalidProducerEpoch` exception was not properly handled on Streams side in all cases. We should catch it and rethrow as TaskMigrated to trigger exception, similar to ProducerFenced.