ekaf is not honoring broker's partition change

It appears from our production environment that broker's partition change is not being honored in ekaf and we are seeing

```
Produce request with correlation id 2628 from client ekaf on partition [layer-insights,0] failed due to Leader not local for partition [layer-insights,0] on broker 2 (kafka.server.KafkaApis)
```

From kafka protocol:

```
The client does not need to keep polling to see if the cluster has changed; it can fetch metadata once when it is instantiated cache that metadata until it receives an error indicating that the metadata is out of date. This error can come in two forms: (1) a socket error indicating the client cannot communicate with a particular broker, (2) an error code in the response to a request indicating that this broker no longer hosts the partition for which data was requested.
```

seems like (2) (invalid partition error code) is not handled in ekaf.