Bump org.apache.kafka:kafka-clients from 3.8.0 to 4.0.0 (#522)

* Bump org.apache.kafka:kafka-clients from 3.8.0 to 4.0.0

Bumps org.apache.kafka:kafka-clients from 3.8.0 to 4.0.0.

---
updated-dependencies:
- dependency-name: org.apache.kafka:kafka-clients
  dependency-version: 4.0.0
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <[email protected]>

* fix deprecations

---------

Signed-off-by: dependabot[bot] <[email protected]>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Brandon Arp <[email protected]>

Author: dependabot[bot]

pom.xml

@@ -85,7 +85,7 @@
     <javassist.version>3.30.2-GA</javassist.version>
     <javassist.maven.core.version>0.2.9</javassist.maven.core.version>
     <jsr305.version>3.0.2</jsr305.version>
-    <kafka.client.version>3.8.0</kafka.client.version>
+    <kafka.client.version>4.0.0</kafka.client.version>
     <logback.version>1.5.18</logback.version>
     <logback.steno.version>2.1.2</logback.steno.version>
     <log4j.over.slf4j.version>2.0.17</log4j.over.slf4j.version>

src/test/java/com/arpnetworking/metrics/common/sources/KafkaSourceTest.java

@@ -24,11 +24,12 @@
 import com.arpnetworking.test.CollectorPeriodicMetrics;
 import com.arpnetworking.test.StringToRecordParser;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Maps;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.MockConsumer;
-import org.apache.kafka.clients.consumer.OffsetResetStrategy;
+import org.apache.kafka.clients.consumer.internals.AutoOffsetResetStrategy;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.TopicPartition;
 import org.junit.After;
@@ -327,7 +328,7 @@ private static List<String> createValues(final String prefix, final int num) {
 
     private static MockConsumer<String, String> createMockConsumer(
             final ConsumerRecords<String, String> consumerRecords) {
-        final MockConsumer<String, String> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
+        final MockConsumer<String, String> consumer = new MockConsumer<>(AutoOffsetResetStrategy.StrategyType.EARLIEST.toString());
         consumer.assign(Collections.singletonList(new TopicPartition(TOPIC, PARTITION)));
         consumer.updateBeginningOffsets(Collections.singletonMap(new TopicPartition(TOPIC, PARTITION), 0L));
         for (final ConsumerRecord<String, String> record : consumerRecords) {
@@ -342,7 +343,9 @@ private static ConsumerRecords<String, String> expectedConsumerRecords() {
         for (final String value : EXPECTED) {
             records.add(new ConsumerRecord<>(TOPIC, PARTITION, offset++, "" + offset, value));
         }
-        return new ConsumerRecords<>(Collections.singletonMap(new TopicPartition(TOPIC, PARTITION), records));
+        return new ConsumerRecords<>(
+                Collections.singletonMap(new TopicPartition(TOPIC, PARTITION), records),
+                Maps.newHashMap());
     }
 
     private static class FillingBlockingQueue extends ArrayBlockingQueue<String> {