Use java generics with Kafka API.
Clear s3_uploads counter.
Returns a copy of topic partitions collections in order to avoid
concurrent modification exception.