Add retry logic to PartitionConsumer

Also fixes initialisation of LeaderConnector.