diff --git a/pom.xml b/pom.xml
index 35dc1add..8917eb58 100644
--- a/pom.xml
+++ b/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>
diff --git a/src/test/java/com/arpnetworking/metrics/common/sources/KafkaSourceTest.java b/src/test/java/com/arpnetworking/metrics/common/sources/KafkaSourceTest.java
index 69cb6ace..ec9958d1 100644
--- a/src/test/java/com/arpnetworking/metrics/common/sources/KafkaSourceTest.java
+++ b/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> {