Repository: apache/streampipes
Pull request #2761 (2024-04-23 08:17:37)

refactor: minor code & logging improvements
<!--
  ~ Licensed to the Apache Software Foundation (ASF) under one or more
  ~ contributor license agreements.  See the NOTICE file distributed with
  ~ this work for additional information regarding copyright ownership.
  ~ The ASF licenses this file to You under the Apache License, Version 2.0
  ~ (the "License"); you may not use this file except in compliance with
  ~ the License.  You may obtain a copy of the License at
  ~
  ~    http://www.apache.org/licenses/LICENSE-2.0
  ~
  ~ Unless required by applicable law or agreed to in writing, software
  ~ distributed under the License is distributed on an "AS IS" BASIS,
  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  ~ See the License for the specific language governing permissions and
  ~ limitations under the License.
  ~
  -->
  
  <!--
Thanks for contributing! Here are some tips you can follow to help us incorporate your contribution quickly and easily:
1. If this is your first time, please read our contributor guidelines:
    - https://streampipes.apache.org/getinvolved.html
    - https://cwiki.apache.org/confluence/display/STREAMPIPES/Getting+Started
2. Make sure the PR title is formatted like: `[#<GitHub issue id>] PR title ...`
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., `[WIP][#<GitHub issue id>] PR title ...`.
4. Please write your PR title to summarize what this PR proposes/fixes.
5. Link the PR to the corresponding GitHub issue (if present) in the `Development` section in the right menu bar. 
6. Be sure to keep the PR description updated to reflect all changes.
7. If possible, provide a concise example to reproduce the issue for a faster review.
8. Make sure tests pass via `mvn clean install`.
9. (Optional) If the contribution is large, please file an Apache ICLA
    - http://apache.org/licenses/icla.pdf
-->

### Purpose
<!--
Please clarify what changes you are proposing and describe how those changes will address the issue.
Furthermore, describe potential consequences the changes might have.
-->

### Remarks
<!--
Is there anything left we need to pay attention on?
Are there some references that might be important? E.g. links to Confluence, or discussions
on the mailing list or GitHub.
-->
PR introduces (a) breaking change(s): no

PR introduces (a) deprecation(s): no


Patch:
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeInfluxQueryBuilder.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeInfluxQueryBuilder.java
index d2f518246b..8c968ca885 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeInfluxQueryBuilder.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeInfluxQueryBuilder.java
@@ -170,7 +170,7 @@ public class DataLakeInfluxQueryBuilder implements IDataLakeQueryBuilder<Query>
   public IDataLakeQueryBuilder<Query> withInclusiveFilter(List<FilterCondition> filterConditions) {
     List<ConjunctionClause> and = new ArrayList<>();
     filterConditions.forEach(c -> and
-        .add(new AndConjunction(new SimpleClause(c.getField(), c.getOperator(), c.getCondition()))));
+        .add(new AndConjunction(new SimpleClause(c.field(), c.operator(), c.condition()))));
 
     addNestedWhereClause(and);
 
@@ -251,7 +251,7 @@ public class DataLakeInfluxQueryBuilder implements IDataLakeQueryBuilder<Query>
         this.selectionQuery.from(env.getTsStorageBucket().getValueOrDefault(), escapeIndex(measurementId));
     this.whereClauses.forEach(selectQuery::where);
 
-    if (this.groupByClauses.size() > 0) {
+    if (!this.groupByClauses.isEmpty()) {
       selectQuery.groupBy(this.groupByClauses.toArray());
     }
 
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/AutoAggregationHandler.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/AutoAggregationHandler.java
index 68bf23171a..071d7a64d2 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/AutoAggregationHandler.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/AutoAggregationHandler.java
@@ -51,8 +51,10 @@ public class AutoAggregationHandler {
   private static final String TIMESTAMP_FIELD = "time";
   private static final String COMMA = ",";
 
-  private final SimpleDateFormat dateFormat1 = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
-  private final SimpleDateFormat dateFormat2 = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
+  // Date format for ISO 8601 timestamps with milliseconds
+  private final SimpleDateFormat isoFormatWithMillis = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
+  // Date format for ISO 8601 timestamps without milliseconds
+  private final SimpleDateFormat isoFormatOnlySeconds = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
 
   private final IDataExplorerQueryManagement dataLakeQueryManagement;
   private final ProvidedRestQueryParams queryParams;
@@ -91,7 +93,7 @@ public class AutoAggregationHandler {
         return disableAutoAgg(this.queryParams);
       }
     } catch (ParseException e) {
-      e.printStackTrace();
+      LOG.error("Parsing of timestamp failed during auto aggregation of query parameters: {}", e.getMessage());
     }
     return null;
   }
@@ -144,7 +146,7 @@ public class AutoAggregationHandler {
     List<SelectColumn> columns =
         Arrays.stream(rawQuery.split(COMMA))
               .map(SelectColumn::fromApiQueryString)
-              .collect(Collectors.toList());
+              .toList();
     return columns.stream()
                   .map(SelectColumn::getOriginalField)
                   .collect(Collectors.joining(COMMA));
@@ -172,9 +174,9 @@ public class AutoAggregationHandler {
 
   private Date tryParseDate(String v) throws ParseException {
     try {
-      return dateFormat1.parse(v);
+      return isoFormatWithMillis.parse(v);
     } catch (ParseException e) {
-      return dateFormat2.parse(v);
+      return isoFormatOnlySeconds.parse(v);
     }
   }
 }
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQueryExecutor.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQueryExecutor.java
index c28c949780..205046303a 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQueryExecutor.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQueryExecutor.java
@@ -95,12 +95,12 @@ public abstract class DataExplorerQueryExecutor<X, W> {
   public SpQueryResult executeQuery(X query,
                                     boolean ignoreMissingValues) {
     if (LOG.isDebugEnabled()) {
-      LOG.debug("Data Lake Query " + asQueryString(query));
+      LOG.debug("Data Lake Query {}", asQueryString(query));
     }
 
     W result = executeQuery(query);
     if (LOG.isDebugEnabled()) {
-      LOG.debug("Data Lake Query Result: " + result.toString());
+      LOG.debug("Data Lake Query Result: {}", result.toString());
     }
 
     return postQuery(result, ignoreMissingValues);
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredCsvOutputWriter.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredCsvOutputWriter.java
index 03ca75530c..3d37ff9230 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredCsvOutputWriter.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredCsvOutputWriter.java
@@ -19,7 +19,7 @@
 package org.apache.streampipes.dataexplorer.query.writer;
 
 import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParams;
-import org.apache.streampipes.dataexplorer.query.writer.item.CsvItemWriter;
+import org.apache.streampipes.dataexplorer.query.writer.item.CsvItemGenerator;
 
 import java.io.IOException;
 import java.io.OutputStream;
@@ -34,7 +34,7 @@ public class ConfiguredCsvOutputWriter extends ConfiguredOutputWriter {
   private static final String COMMA = ",";
   private static final String SEMICOLON = ";";
 
-  private CsvItemWriter itemWriter;
+  private CsvItemGenerator itemGenerator;
   private String delimiter = COMMA;
 
   @Override
@@ -43,7 +43,7 @@ public class ConfiguredCsvOutputWriter extends ConfiguredOutputWriter {
     if (params.has(QP_CSV_DELIMITER)) {
       delimiter = params.getAsString(QP_CSV_DELIMITER).equals("comma") ? COMMA : SEMICOLON;
     }
-    this.itemWriter = new CsvItemWriter(delimiter);
+    this.itemGenerator = new CsvItemGenerator(delimiter);
   }
 
   @Override
@@ -65,7 +65,7 @@ public class ConfiguredCsvOutputWriter extends ConfiguredOutputWriter {
       outputStream.write(toBytes(makeHeaderLine(columnNames)));
     }
 
-    outputStream.write(toBytes(itemWriter.createItem(row, columnNames) + LINE_SEPARATOR));
+    outputStream.write(toBytes(itemGenerator.createItem(row, columnNames) + LINE_SEPARATOR));
   }
 
   private String makeHeaderLine(List<String> columns) {
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredJsonOutputWriter.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredJsonOutputWriter.java
index 0781c7334a..e1311c9e47 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredJsonOutputWriter.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredJsonOutputWriter.java
@@ -20,7 +20,7 @@ package org.apache.streampipes.dataexplorer.query.writer;
 
 import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParams;
 import org.apache.streampipes.dataexplorer.query.writer.item.ItemGenerator;
-import org.apache.streampipes.dataexplorer.query.writer.item.JsonItemWriter;
+import org.apache.streampipes.dataexplorer.query.writer.item.JsonItemGenerator;
 
 import com.google.gson.Gson;
 
@@ -33,11 +33,11 @@ public class ConfiguredJsonOutputWriter extends ConfiguredOutputWriter {
   private static final String BEGIN_ARRAY = "[";
   private static final String END_ARRAY = "]";
 
-  private final ItemGenerator jsonObjectWriter;
+  private final ItemGenerator itemGenerator;
 
   public ConfiguredJsonOutputWriter() {
     Gson gson = new Gson();
-    this.jsonObjectWriter = new JsonItemWriter(gson);
+    this.itemGenerator = new JsonItemGenerator(gson);
   }
 
   @Override
@@ -65,7 +65,7 @@ public class ConfiguredJsonOutputWriter extends ConfiguredOutputWriter {
       outputStream.write(toBytes(","));
     }
 
-    var item = jsonObjectWriter.createItem(row, columnNames);
+    var item = itemGenerator.createItem(row, columnNames);
     outputStream.write(toBytes(item));
   }
 }
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/CsvItemWriter.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/CsvItemGenerator.java
similarity index 91%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/CsvItemWriter.java
rename to streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/CsvItemGenerator.java
index bd80f2d9ab..f509338d3c 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/CsvItemWriter.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/CsvItemGenerator.java
@@ -19,9 +19,9 @@
 
 package org.apache.streampipes.dataexplorer.query.writer.item;
 
-public class CsvItemWriter extends ItemGenerator {
+public class CsvItemGenerator extends ItemGenerator {
 
-  public CsvItemWriter(String delimiter) {
+  public CsvItemGenerator(String delimiter) {
     super(delimiter);
   }
 
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/JsonItemWriter.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/JsonItemGenerator.java
similarity index 93%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/JsonItemWriter.java
rename to streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/JsonItemGenerator.java
index fde43721d4..e135bb8cdf 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/JsonItemWriter.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/JsonItemGenerator.java
@@ -21,14 +21,14 @@ package org.apache.streampipes.dataexplorer.query.writer.item;
 
 import com.google.gson.Gson;
 
-public class JsonItemWriter extends ItemGenerator {
+public class JsonItemGenerator extends ItemGenerator {
 
   private static final String BEGIN_OBJECT = "{";
   private static final String END_OBJECT = "}";
 
   private final Gson gson;
 
-  public JsonItemWriter(Gson gson) {
+  public JsonItemGenerator(Gson gson) {
     super(COMMA_SEPARATOR);
     this.gson = gson;
   }
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/FilterCondition.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/FilterCondition.java
index 396bed21d2..ed2592d8d3 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/FilterCondition.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/FilterCondition.java
@@ -17,28 +17,5 @@
  */
 package org.apache.streampipes.dataexplorer.querybuilder;
 
-public class FilterCondition {
-
-  private String field;
-  private String operator;
-  private Object condition;
-
-  public FilterCondition(String field, String operator, Object condition) {
-    this.field = field;
-    this.operator = operator;
-    this.condition = condition;
-  }
-
-  public String getField() {
-    return field;
-  }
-
-  public String getOperator() {
-    return operator;
-  }
-
-  public Object getCondition() {
-    return condition;
-  }
-
+public record FilterCondition(String field, String operator, Object condition) {
 }
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java
index cdecf2a56d..6f6a0477ae 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/Oi4Adapter.java
@@ -37,13 +37,13 @@ import org.apache.streampipes.extensions.management.connect.adapter.parser.json.
 import org.apache.streampipes.messaging.InternalEventProcessor;
 import org.apache.streampipes.model.AdapterType;
 import org.apache.streampipes.model.connect.guess.GuessSchema;
+import org.apache.streampipes.model.extensions.ExtensionAssetType;
 import org.apache.streampipes.model.schema.EventSchema;
 import org.apache.streampipes.sdk.StaticProperties;
 import org.apache.streampipes.sdk.builder.adapter.AdapterConfigurationBuilder;
 import org.apache.streampipes.sdk.helpers.Alternatives;
 import org.apache.streampipes.sdk.helpers.Labels;
 import org.apache.streampipes.sdk.helpers.Locales;
-import org.apache.streampipes.sdk.utils.Assets;
 import org.apache.streampipes.sdk.utils.Datatypes;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -60,7 +60,6 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -87,7 +86,6 @@ public class Oi4Adapter implements StreamPipesAdapter {
 
   private List<String> selectedSensors;
   private String givenSensorType;
-  private String appId;
   protected final ObjectMapper mapper;
 
 
@@ -101,7 +99,7 @@ public class Oi4Adapter implements StreamPipesAdapter {
     return AdapterConfigurationBuilder
         .create(ID, 0, Oi4Adapter::new)
         .withLocales(Locales.EN)
-        .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+        .withAssets(ExtensionAssetType.DOCUMENTATION, ExtensionAssetType.ICON)
         .withCategory(AdapterType.Generic, AdapterType.Manufacturing)
         .requiredTextParameter(MqttConnectUtils.getBrokerUrlLabel())
         .requiredAlternatives(MqttConnectUtils.getAccessModeLabel(), MqttConnectUtils.getAlternativesOne(),
@@ -174,7 +172,7 @@ public class Oi4Adapter implements StreamPipesAdapter {
         OI4AdapterLabels.LABEL_KEY_SENSORS_ALTERNATIVES
     );
 
-    appId = extractor.textParameter(OI4AdapterLabels.LABEL_KEY_APP_ID);
+    String appId = extractor.textParameter(OI4AdapterLabels.LABEL_KEY_APP_ID);
 
     if (selectedAlternativeSelectedSensors.equals(OI4AdapterLabels.LABEL_KEY_SENSORS_ALL_ALTERNATIVE)) {
       // An empty list is used to indicated that messages from all sensors should be collected
@@ -197,7 +195,7 @@ public class Oi4Adapter implements StreamPipesAdapter {
   public void onAdapterStopped(
       IAdapterParameterExtractor extractor,
       IAdapterRuntimeContext adapterRuntimeContext
-  ) throws AdapterException {
+  ) {
     mqttConsumer.close();
   }
 
@@ -311,40 +309,42 @@ public class Oi4Adapter implements StreamPipesAdapter {
 
   private Map<String, Object> extractPayload(NetworkMessage message) throws ParseException {
 
-    var dataMessageOpt = findProcessDataInputMessage(message);
+    var dataMessages = findProcessDataInputMessage(message);
 
-    if (dataMessageOpt.isPresent()) {
+    if (!dataMessages.isEmpty()) {
 
-      var dataMessage = dataMessageOpt.get();
+      for (var dataMessage: dataMessages) {
 
-      var sensorId = getSensorIdFromSource(dataMessage.source());
+        var sensorId = getSensorIdFromSource(dataMessage.source());
 
-      // Verify that the message corresponds to the designated sensor type.
-      // This validation relies on the assumption that the source information includes the sensor type.
-      if (dataMessage.source()
-          .contains(givenSensorType)) {
+        // Verify that the message corresponds to the designated sensor type.
+        // This validation relies on the assumption that the source information includes the sensor type.
+        if (dataMessage.source()
+                       .contains(givenSensorType)) {
 
-        // an empty list of selected sensors means that we want to collect data from all sensors available
-        if (selectedSensors.isEmpty() || selectedSensors.contains(sensorId)) {
+          // an empty list of selected sensors means that we want to collect data from all sensors available
+          if (selectedSensors.isEmpty() || selectedSensors.contains(sensorId)) {
 
-          return extractAndEnrichMessagePayload(dataMessage, sensorId);
+            return extractAndEnrichMessagePayload(dataMessage, sensorId);
+          }
         }
       }
     }
     throw new ParseException("No process data message");
   }
 
-  private Optional<DataSetMessage> findProcessDataInputMessage(NetworkMessage message) {
+  private List<DataSetMessage> findProcessDataInputMessage(NetworkMessage message) {
     return message.messages()
-        .stream()
-        .filter(msg -> msg.filter()
-            .equals(OI4AdapterLabels.MESSAGE_VALUE_FILTER))
-        .findFirst();
+                  .stream()
+                  .filter(msg -> msg.filter()
+                  .equals(OI4AdapterLabels.MESSAGE_VALUE_FILTER))
+                  .toList();
   }
 
   private Map<String, Object> extractAndEnrichMessagePayload(DataSetMessage dataSetMessage, String sensorId) {
     var payload = dataSetMessage
-        .payload();
+        .payload()
+        .get(0);
     try {
       payload.put(
           OI4AdapterLabels.EVENT_KEY_TIMESTAMP,
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/model/DataSetMessage.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/model/DataSetMessage.java
index 6fb62b3300..f50dd2f76a 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/model/DataSetMessage.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/oi4/model/DataSetMessage.java
@@ -21,6 +21,7 @@ package org.apache.streampipes.connect.iiot.adapters.oi4.model;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -36,7 +37,7 @@ public record DataSetMessage(
     @JsonProperty("Timestamp") String timestamp,
     @JsonProperty("Filter") String filter,
     @JsonProperty("Source") String source,
-    @JsonProperty("Payload") Map<String, Object> payload
+    @JsonProperty("Payload") List<Map<String, Object>> payload
 ) {
 
 }


Test patch:
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/param/WhereStatementParamsTest.java b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/param/WhereStatementParamsTest.java
index 6859a41531..984aee2bed 100644
--- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/param/WhereStatementParamsTest.java
+++ b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/param/WhereStatementParamsTest.java
@@ -53,9 +53,9 @@ public class WhereStatementParamsTest {
   private void assertWhereCondition(WhereClauseParams result, FilterCondition expected) {
     assertEquals(1, result.getWhereConditions().size());
     FilterCondition resultingFilterCondition = result.getWhereConditions().get(0);
-    assertEquals(expected.getField(), resultingFilterCondition.getField());
-    assertEquals(expected.getOperator(), resultingFilterCondition.getOperator());
-    assertEquals(expected.getCondition(), resultingFilterCondition.getCondition());
+    assertEquals(expected.field(), resultingFilterCondition.field());
+    assertEquals(expected.operator(), resultingFilterCondition.operator());
+    assertEquals(expected.condition(), resultingFilterCondition.condition());
   }
 
 
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestCsvItemWriter.java b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestCsvItemGenerator.java
similarity index 90%
rename from streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestCsvItemWriter.java
rename to streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestCsvItemGenerator.java
index bb45358a7c..b24adf5838 100644
--- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestCsvItemWriter.java
+++ b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestCsvItemGenerator.java
@@ -22,14 +22,14 @@ import org.junit.jupiter.api.Test;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
-public class TestCsvItemWriter extends TestItemWriter {
+public class TestCsvItemGenerator extends TestItemGenerator {
 
   private static final String ExpectedComma = "1668578077051,test,1";
   private static final String ExpectedSemicolon = "1668578077051;test;1";
 
   @Test
   public void testCsvItemWriterCommaSeparated() {
-    var writer = new CsvItemWriter(",");
+    var writer = new CsvItemGenerator(",");
 
     String result = writer.createItem(row, columns);
 
@@ -38,7 +38,7 @@ public class TestCsvItemWriter extends TestItemWriter {
 
   @Test
   public void testCsvItemWriterSemicolonSeparated() {
-    var writer = new CsvItemWriter(";");
+    var writer = new CsvItemGenerator(";");
 
     String result = writer.createItem(row, columns);
 
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestItemWriter.java b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestItemGenerator.java
similarity index 96%
rename from streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestItemWriter.java
rename to streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestItemGenerator.java
index 0c113ca78d..3dd0fd7672 100644
--- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestItemWriter.java
+++ b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestItemGenerator.java
@@ -23,7 +23,7 @@ import org.junit.jupiter.api.BeforeEach;
 import java.util.Arrays;
 import java.util.List;
 
-public abstract class TestItemWriter {
+public abstract class TestItemGenerator {
 
   protected List<Object> row;
   protected List<String> columns;
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestJsonItemWriter.java b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestJsonItemGenerator.java
similarity index 91%
rename from streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestJsonItemWriter.java
rename to streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestJsonItemGenerator.java
index 518194b4d4..2b3dcb58b1 100644
--- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestJsonItemWriter.java
+++ b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestJsonItemGenerator.java
@@ -23,13 +23,13 @@ import org.junit.jupiter.api.Test;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
-public class TestJsonItemWriter extends TestItemWriter {
+public class TestJsonItemGenerator extends TestItemGenerator {
 
   private static final String Expected = "{\"time\": 1668578077051,\"string\": \"test\",\"number\": 1}";
 
   @Test
   public void testJsonWriter() {
-    var writer = new JsonItemWriter(new Gson());
+    var writer = new JsonItemGenerator(new Gson());
 
     String result = writer.createItem(row, columns);