Backport of https://github.com/spring-projects/spring-kafka/issues/4303
Backport of #4303