Align Flink Kafka config with others

theodolite-benchmarks/flink-commons/src/main/java/rocks/theodolite/benchmarks/commons/flink/KafkaConnectorFactory.java

@@ -23,6 +23,8 @@ import rocks.theodolite.benchmarks.commons.flink.util.SerializableSupplier;
  */
 public class KafkaConnectorFactory {
 
+  private static final String AUTO_OFFSET_RESET_EARLIEST = "earliest";
+
   private static final Duration PRODUCER_TRANSACTION_TIMEOUT = Duration.ofMinutes(5);
 
   private final Properties kafkaProps = new Properties();
@@ -50,7 +52,7 @@ public class KafkaConnectorFactory {
   public <T> FlinkKafkaConsumer<T> createConsumer(final String topic,
       final DeserializationSchema<T> deserializationSchema) {
     return this.createBaseConsumer(
-        new FlinkKafkaConsumer<>(topic, deserializationSchema, this.cloneProperties()));
+        new FlinkKafkaConsumer<>(topic, deserializationSchema, this.buildConsumerProperties()));
   }
 
   /**
@@ -60,7 +62,7 @@ public class KafkaConnectorFactory {
   public <T> FlinkKafkaConsumer<T> createConsumer(final String topic,
       final KafkaDeserializationSchema<T> deserializationSchema) {
     return this.createBaseConsumer(
-        new FlinkKafkaConsumer<>(topic, deserializationSchema, this.cloneProperties()));
+        new FlinkKafkaConsumer<>(topic, deserializationSchema, this.buildConsumerProperties()));
   }
 
   /**
@@ -145,6 +147,14 @@ public class KafkaConnectorFactory {
     return producerProps;
   }
 
+  private Properties buildConsumerProperties() {
+    final Properties consumerProps = this.cloneProperties();
+    consumerProps.setProperty(
+        ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
+        AUTO_OFFSET_RESET_EARLIEST);
+    return consumerProps;
+  }
+
   private Properties cloneProperties() {
     final Properties props = new Properties();
     props.putAll(this.kafkaProps);