Merge pull request opensearch-project#456 from gregschohn/BlockingTra…

…fficSourceWithKeepAlives

Blocking traffic source with keep alives

TrafficCapture/nettyWireLogging/src/test/java/org/opensearch/migrations/trafficcapture/netty/ConditionallyReliableLoggingHttpRequestHandlerTest.java

@@ -57,7 +57,7 @@ public CodedOutputStreamAndByteBufferWrapper createStream() {
 
             cos.flush();
             byteBufferAtomicReference.set(osh.getByteBuffer().flip().asReadOnlyBuffer());
-            log.error("byteBufferAtomicReference.get="+byteBufferAtomicReference.get());
+            log.trace("byteBufferAtomicReference.get="+byteBufferAtomicReference.get());
 
             return CompletableFuture.completedFuture(flushCount.incrementAndGet());
         }

TrafficCapture/replayerPlugins/jsonMessageTransformers/jsonJoltMessageTransformerProvider/src/test/java/org/opensearch/migrations/replay/JsonTransformerTest.java

@@ -66,7 +66,7 @@ public void testHttpTransform() throws IOException {
                 .build();
         var transformedDocument = transformer.transformJson(documentJson);
         String transformedJsonOutputStr = emitJson(transformedDocument);
-        log.error("transformed json document: "+transformedJsonOutputStr);
+        log.info("transformed json document: "+transformedJsonOutputStr);
         Assertions.assertTrue(transformedJsonOutputStr.contains(DUMMY_HOSTNAME_TEST_STRING));
     }
 }

TrafficCapture/trafficCaptureProxyServer/src/test/resources/log4j2.properties

@@ -6,6 +6,6 @@ appender.console.target = SYSTEM_ERR
 appender.console.layout.type = PatternLayout
 appender.console.layout.pattern = [%-5level] %d{yyyy-MM-dd HH:mm:ss,SSS}{UTC} [%t] %c{1} - %msg%equals{ ctx=%mdc}{ ctx=\{\}}{}%n
 
-rootLogger.level = trace
+rootLogger.level = debug
 rootLogger.appenderRefs = stderr
 rootLogger.appenderRef.stderr.ref = STDERR

TrafficCapture/trafficReplayer/src/main/java/org/opensearch/migrations/replay/TrafficCaptureSourceFactory.java

@@ -2,13 +2,14 @@
 
 import lombok.extern.slf4j.Slf4j;
 import org.opensearch.migrations.replay.kafka.KafkaBehavioralPolicy;
-import org.opensearch.migrations.replay.kafka.KafkaProtobufConsumer;
+import org.opensearch.migrations.replay.kafka.KafkaTrafficCaptureSource;
 import org.opensearch.migrations.replay.traffic.source.BlockingTrafficSource;
 import org.opensearch.migrations.replay.traffic.source.ISimpleTrafficCaptureSource;
 import org.opensearch.migrations.replay.traffic.source.InputStreamOfTraffic;
 
 import java.io.FileInputStream;
 import java.io.IOException;
+import java.time.Clock;
 import java.time.Duration;
 
 @Slf4j
@@ -31,8 +32,9 @@ private TrafficCaptureSourceFactory() {}
         }
 
         if (isKafkaActive) {
-            return KafkaProtobufConsumer.buildKafkaConsumer(appParams.kafkaTrafficBrokers, appParams.kafkaTrafficTopic,
-                appParams.kafkaTrafficGroupId, appParams.kafkaTrafficEnableMSKAuth, appParams.kafkaTrafficPropertyFile, new KafkaBehavioralPolicy());
+            return KafkaTrafficCaptureSource.buildKafkaConsumer(appParams.kafkaTrafficBrokers, appParams.kafkaTrafficTopic,
+                appParams.kafkaTrafficGroupId, appParams.kafkaTrafficEnableMSKAuth, appParams.kafkaTrafficPropertyFile,
+                    Clock.systemUTC(), new KafkaBehavioralPolicy());
         } else if (isInputFileActive) {
             return new InputStreamOfTraffic(new FileInputStream(appParams.inputFilename));
         } else {

TrafficCapture/trafficReplayer/src/main/java/org/opensearch/migrations/replay/datatypes/PojoTrafficStreamKey.java

@@ -1,10 +1,12 @@
 package org.opensearch.migrations.replay.datatypes;
 
+import lombok.EqualsAndHashCode;
 import lombok.ToString;
 import org.opensearch.migrations.trafficcapture.protos.TrafficStream;
 import org.opensearch.migrations.trafficcapture.protos.TrafficStreamUtils;
 
 @ToString
+@EqualsAndHashCode()
 public class PojoTrafficStreamKey implements ITrafficStreamKey {
     private final String nodeId;
     private final String connectionId;

TrafficCapture/trafficReplayer/src/main/java/org/opensearch/migrations/replay/kafka/KafkaCommitOffsetData.java

@@ -0,0 +1,8 @@
+package org.opensearch.migrations.replay.kafka;
+
+public interface KafkaCommitOffsetData {
+    int getPartition();
+    long getOffset();
+    int getGeneration();
+
+}