refactor(#2964): timestamp conversion is broken in adapters (#2986)

* fix(#2964): Fix test valueRules.spec.ts

* chore(#2964): Refactor FileReplayAdapter and add unit tests

* refactor(#2964): Fix timestamp transformations within FileReplayAdapter

* refactor(#2964): Refactor method onAdapterStarted in FileReplayAdapter

* refactor(#2964): Fix e2e test for file replay adapter

* refactor(#2964): Throw exception when a user selects the add timestamp rule with the file replay adapter

* refactor(#2964): Fix checkstyle

* refactor(#2964): Add tests for number timestamp conversion

* refactor(#2964): Fix linting error

* refactor(#2964): Fix checkstyle

* refactor(#2964): Add time zone to TimestampTransformationRule

* refactor(#2964): Fix time zone bug in timestamp rule

* refactor(#2964): Fix e2e tests

* refactor(#2964): Fix linting problem

streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/SupportsNestedTransformationRule.java

@@ -18,7 +18,7 @@
 
 package org.apache.streampipes.connect.shared.preprocessing;
 
-import org.apache.streampipes.connect.shared.preprocessing.transform.TransformationRule;
+import org.apache.streampipes.extensions.api.connect.TransformationRule;
 
 import java.util.List;
 import java.util.Map;

streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/elements/AdapterTransformationPipelineElement.java

@@ -19,9 +19,9 @@
 package org.apache.streampipes.connect.shared.preprocessing.elements;
 
 import org.apache.streampipes.connect.shared.preprocessing.generator.TransformationRuleGeneratorVisitor;
-import org.apache.streampipes.connect.shared.preprocessing.transform.TransformationRule;
 import org.apache.streampipes.connect.shared.preprocessing.utils.Utils;
 import org.apache.streampipes.extensions.api.connect.IAdapterPipelineElement;
+import org.apache.streampipes.extensions.api.connect.TransformationRule;
 import org.apache.streampipes.model.connect.rules.TransformationRuleDescription;
 
 import java.util.List;

streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/generator/TransformationRuleGeneratorVisitor.java

@@ -18,7 +18,7 @@
 
 package org.apache.streampipes.connect.shared.preprocessing.generator;
 
-import org.apache.streampipes.connect.shared.preprocessing.transform.TransformationRule;
+import org.apache.streampipes.extensions.api.connect.TransformationRule;
 import org.apache.streampipes.model.connect.rules.ITransformationRuleVisitor;
 
 import java.util.ArrayList;

streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/transform/schema/AddValueTransformationRule.java

@@ -19,7 +19,7 @@
 package org.apache.streampipes.connect.shared.preprocessing.transform.schema;
 
 import org.apache.streampipes.connect.shared.DatatypeUtils;
-import org.apache.streampipes.connect.shared.preprocessing.transform.TransformationRule;
+import org.apache.streampipes.extensions.api.connect.TransformationRule;
 
 import java.util.Map;
 

streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/transform/schema/MoveTransformationRule.java

@@ -18,7 +18,7 @@
 
 package org.apache.streampipes.connect.shared.preprocessing.transform.schema;
 
-import org.apache.streampipes.connect.shared.preprocessing.transform.TransformationRule;
+import org.apache.streampipes.extensions.api.connect.TransformationRule;
 
 import java.util.HashMap;
 import java.util.List;

streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/transform/stream/DuplicateFilterPipelineElement.java

@@ -18,7 +18,7 @@
 
 package org.apache.streampipes.connect.shared.preprocessing.transform.stream;
 
-import org.apache.streampipes.connect.shared.preprocessing.transform.TransformationRule;
+import org.apache.streampipes.extensions.api.connect.TransformationRule;
 
 import java.util.HashMap;
 import java.util.Map;

streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/transform/stream/EventRateTransformationRule.java

@@ -18,7 +18,7 @@
 
 package org.apache.streampipes.connect.shared.preprocessing.transform.stream;
 
-import org.apache.streampipes.connect.shared.preprocessing.transform.TransformationRule;
+import org.apache.streampipes.extensions.api.connect.TransformationRule;
 
 import java.util.Map;
 

streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/transform/value/AddTimestampTransformationRule.java

@@ -18,7 +18,7 @@
 
 package org.apache.streampipes.connect.shared.preprocessing.transform.value;
 
-import org.apache.streampipes.connect.shared.preprocessing.transform.TransformationRule;
+import org.apache.streampipes.extensions.api.connect.TransformationRule;
 
 import java.util.Map;
 

streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/transform/value/DatatypeTransformationRule.java

@@ -20,7 +20,7 @@
 
 
 import org.apache.streampipes.connect.shared.DatatypeUtils;
-import org.apache.streampipes.connect.shared.preprocessing.transform.TransformationRule;
+import org.apache.streampipes.extensions.api.connect.TransformationRule;
 
 import java.util.Map;
 

streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/transform/value/TimestampTransformationRule.java

@@ -28,6 +28,8 @@
 import java.util.List;
 import java.util.Map;
 
+import static java.util.TimeZone.getTimeZone;
+
 public class TimestampTransformationRule extends SupportsNestedTransformationRule {
 
   private final List<String> eventKey;
@@ -48,6 +50,7 @@ public TimestampTransformationRule(List<String> eventKey,
 
     if (mode == TimestampTranformationRuleMode.FORMAT_STRING) {
       dateFormatter = new SimpleDateFormat(formatString);
+      dateFormatter.setTimeZone(getTimeZone("UTC"));
     }
   }
 

streampipes-connect-shared/src/test/java/org/apache/streampipes/connect/shared/preprocessing/transform/schema/SchemaEventTransformerTest.java

@@ -18,7 +18,7 @@
 
 package org.apache.streampipes.connect.shared.preprocessing.transform.schema;
 
-import org.apache.streampipes.connect.shared.preprocessing.transform.TransformationRule;
+import org.apache.streampipes.extensions.api.connect.TransformationRule;
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;

streampipes-connect-shared/src/test/java/org/apache/streampipes/connect/shared/preprocessing/transform/value/ValueEventTransformerTest.java

@@ -18,7 +18,7 @@
 
 package org.apache.streampipes.connect.shared.preprocessing.transform.value;
 
-import org.apache.streampipes.connect.shared.preprocessing.transform.TransformationRule;
+import org.apache.streampipes.extensions.api.connect.TransformationRule;
 import org.apache.streampipes.model.schema.EventProperty;
 import org.apache.streampipes.model.schema.EventPropertyPrimitive;
 import org.apache.streampipes.model.schema.EventSchema;

streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/StreamPipesAdapter.java

@@ -22,11 +22,26 @@
 import org.apache.streampipes.extensions.api.connect.context.IAdapterGuessSchemaContext;
 import org.apache.streampipes.extensions.api.connect.context.IAdapterRuntimeContext;
 import org.apache.streampipes.extensions.api.extractor.IAdapterParameterExtractor;
+import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 import org.apache.streampipes.model.connect.guess.GuessSchema;
 
 public interface StreamPipesAdapter {
   IAdapterConfiguration declareConfig();
 
+  /**
+   * Preprocesses the adapter description before the adapter is invoked.
+   *
+   * <p>This method is designed to allow adapters to modify the adapter description prior to invocation.
+   * It is particularly useful for adapters that need to manipulate certain values internally,
+   * e.g. bypassing the adapter preprocessing pipeline. An example of such an adapter is the FileReplayAdapter,
+   * which manipulates timestamp values.</p>
+   *
+   * <p>This is a default method and does not need to be overridden unless specific preprocessing is required.</p>
+   *
+   * @param adapterDescription The adapter description to be preprocessed.
+   */
+  default void preprocessAdapterDescription(AdapterDescription adapterDescription) {};
+
   void onAdapterStarted(IAdapterParameterExtractor extractor,
                         IEventCollector collector,
                         IAdapterRuntimeContext adapterRuntimeContext) throws AdapterException;

streampipes-connect-shared/src/main/java/org/apache/streampipes/connect/shared/preprocessing/transform/TransformationRule.java → streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/TransformationRule.java

@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.connect.shared.preprocessing.transform;
+package org.apache.streampipes.extensions.api.connect;
 
 import java.util.Map;
 

streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/AdapterWorkerManagement.java

@@ -63,6 +63,11 @@ public void invokeAdapter(AdapterDescription adapterDescription) throws AdapterE
           newAdapterInstance,
           adapterDescription);
 
+      // This method allows adapters to modify the adapter description prior to invocation.
+      // It is particularly useful for adapters like FileReplayAdapter that need to manipulate timestamp values
+      // internally, bypassing the adapter preprocessing pipeline.
+      newAdapterInstance.preprocessAdapterDescription(adapterDescription);
+
       var registeredParsers = newAdapterInstance.declareConfig().getSupportedParsers();
       var extractor = AdapterParameterExtractor.from(adapterDescription, registeredParsers);
       var eventCollector = EventCollector.from(adapterDescription);