Repository: apache/streampipes
Pull request #1386 (2023-03-04 22:22:06)

Remove obsolete features from data explorer module (#1385)
<!--
  ~ 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.
-->

see #1385 

### Breaking Changes
Remove the following unused API endpoints:
* `v4/datalake/configuration` (`POST` & `GET`) 

### 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): yes

PR introduces (a) deprecation(s): no



Patch:
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
index 8be54973ff..92094b2eba 100644
--- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
@@ -22,6 +22,8 @@ public enum Envs {
   SP_HOST("SP_HOST"),
   SP_PORT("SP_PORT"),
 
+  SP_CORE_ASSET_BASE_DIR("SP_CORE_ASSET_BASE_DIR"),
+
   @Deprecated(since = "0.90.0", forRemoval = true)
   SP_CONSUL_LOCATION("CONSUL_LOCATION",
       DefaultEnvValues.CONSUL_HOST_DEFAULT,
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java
index d58260a908..83fd1ebd10 100644
--- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java
@@ -174,6 +174,11 @@ public class DefaultEnvironment implements Environment {
     return new StringEnvironmentVariable(Envs.SP_INITIAL_ADMIN_PASSWORD);
   }
 
+  @Override
+  public StringEnvironmentVariable getCoreAssetBaseDir() {
+    return new StringEnvironmentVariable(Envs.SP_CORE_ASSET_BASE_DIR);
+  }
+
   @Override
   public StringEnvironmentVariable getConsulLocation() {
     return new StringEnvironmentVariable(Envs.SP_CONSUL_LOCATION);
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java
index 2bf5913038..c4a9081a43 100644
--- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java
@@ -100,4 +100,6 @@ public interface Environment {
   StringEnvironmentVariable getInitialAdminEmail();
 
   StringEnvironmentVariable getInitialAdminPassword();
+
+  StringEnvironmentVariable getCoreAssetBaseDir();
 }
diff --git a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java
index caa2c74d8e..d38afc48f8 100644
--- a/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java
+++ b/streampipes-config/src/main/java/org/apache/streampipes/config/backend/BackendConfig.java
@@ -85,10 +85,17 @@ public enum BackendConfig {
   }
 
   private String makeStreamPipesHomeLocation() {
-    return System.getProperty("user.home")
-        + File.separator
-        + ".streampipes"
-        + File.separator;
+    var userDefinedAssetDir = getEnvironment().getCoreAssetBaseDir();
+    var assetDirAppendix = getSpAssetDirAppendix();
+    if (userDefinedAssetDir.exists()) {
+      return userDefinedAssetDir.getValue() + assetDirAppendix;
+    } else {
+      return System.getProperty("user.home") + assetDirAppendix;
+    }
+  }
+
+  private String getSpAssetDirAppendix() {
+    return File.separator + ".streampipes" + File.separator;
   }
 
   public String getJmsHost() {
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
index 6cc0d56206..7f397af7a6 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java
@@ -21,10 +21,7 @@ package org.apache.streampipes.dataexplorer;
 import org.apache.streampipes.commons.environment.Environment;
 import org.apache.streampipes.commons.environment.Environments;
 import org.apache.streampipes.dataexplorer.commons.influx.InfluxClientProvider;
-import org.apache.streampipes.dataexplorer.param.RetentionPolicyQueryParams;
 import org.apache.streampipes.dataexplorer.query.DeleteDataQuery;
-import org.apache.streampipes.dataexplorer.query.EditRetentionPolicyQuery;
-import org.apache.streampipes.dataexplorer.query.ShowRetentionPolicyQuery;
 import org.apache.streampipes.dataexplorer.utils.DataExplorerUtils;
 import org.apache.streampipes.dataexplorer.v4.ProvidedQueryParams;
 import org.apache.streampipes.dataexplorer.v4.params.QueryParamsV4;
@@ -33,9 +30,7 @@ import org.apache.streampipes.dataexplorer.v4.query.QueryResultProvider;
 import org.apache.streampipes.dataexplorer.v4.query.StreamedQueryResultProvider;
 import org.apache.streampipes.dataexplorer.v4.query.writer.OutputFormat;
 import org.apache.streampipes.dataexplorer.v4.utils.DataLakeManagementUtils;
-import org.apache.streampipes.model.datalake.DataLakeConfiguration;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
-import org.apache.streampipes.model.datalake.DataLakeRetentionPolicy;
 import org.apache.streampipes.model.datalake.SpQueryResult;
 import org.apache.streampipes.model.schema.EventProperty;
 import org.apache.streampipes.model.schema.EventPropertyList;
@@ -108,58 +103,12 @@ public class DataLakeManagementV4 {
     return false;
   }
 
-  public SpQueryResult deleteData(String measurementID) {
-    return this.deleteData(measurementID, null, null);
-  }
-
   public SpQueryResult deleteData(String measurementID, Long startDate, Long endDate) {
     Map<String, QueryParamsV4> queryParts =
         DataLakeManagementUtils.getDeleteQueryParams(measurementID, startDate, endDate);
     return new DataExplorerQueryV4(queryParts).executeQuery(true);
   }
 
-  public DataLakeConfiguration getDataLakeConfiguration() {
-    List<DataLakeRetentionPolicy> retentionPolicies = getAllExistingRetentionPolicies();
-    return new DataLakeConfiguration(retentionPolicies);
-  }
-
-  public String editMeasurementConfiguration(DataLakeConfiguration config, boolean resetToDefault) {
-
-    List<DataLakeRetentionPolicy> existingRetentionPolicies = getAllExistingRetentionPolicies();
-
-    if (resetToDefault) {
-      if (existingRetentionPolicies.size() > 1) {
-        String drop =
-            new EditRetentionPolicyQuery(RetentionPolicyQueryParams.from("custom", "0s"), "DROP").executeQuery();
-      }
-      return new EditRetentionPolicyQuery(RetentionPolicyQueryParams.from("autogen", "0s"), "DEFAULT").executeQuery();
-    } else {
-
-      Integer batchSize = config.getBatchSize();
-      Integer flushDuration = config.getFlushDuration();
-
-      //
-      // TODO:
-      // - Implementation of parameter update for batchSize and flushDuration
-      // - Updating multiple retention policies
-      //
-
-      String operation = "CREATE";
-      if (existingRetentionPolicies.size() > 1) {
-        operation = "ALTER";
-      }
-      return new EditRetentionPolicyQuery(RetentionPolicyQueryParams.from("custom", "1d"), operation).executeQuery();
-    }
-  }
-
-  public List<DataLakeRetentionPolicy> getAllExistingRetentionPolicies() {
-    //
-    // TODO:
-    // - Implementation of parameter return for batchSize and flushDuration
-    //
-    return new ShowRetentionPolicyQuery(RetentionPolicyQueryParams.from("", "0s")).executeQuery();
-  }
-
   public boolean removeEventProperty(String measurementID) {
     boolean isSuccess = false;
     CouchDbClient couchDbClient = Utils.getCouchDbDataLakeClient();
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/model/Order.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/model/Order.java
deleted file mode 100644
index e5f27458a3..0000000000
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/model/Order.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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.
- *
- */
-package org.apache.streampipes.dataexplorer.model;
-
-public enum Order {
-
-  ASC("ASC"),
-  DESC("DESC");
-
-  private String value;
-
-  Order(String value) {
-    this.value = value;
-  }
-
-  public String toValue() {
-    return value;
-  }
-}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/QueryParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/QueryParams.java
deleted file mode 100644
index d664eb269b..0000000000
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/QueryParams.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * 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.
- *
- */
-package org.apache.streampipes.dataexplorer.param;
-
-public class QueryParams {
-
-  private final String index;
-
-  protected QueryParams(String index) {
-    this.index = index;
-  }
-
-  public static QueryParams from(String index) {
-    return new QueryParams(index);
-  }
-
-  public String getIndex() {
-    return index;
-  }
-
-}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/RetentionPolicyQueryParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/RetentionPolicyQueryParams.java
deleted file mode 100644
index b42fa6d486..0000000000
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/RetentionPolicyQueryParams.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * 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.
- *
- */
-
-package org.apache.streampipes.dataexplorer.param;
-
-public class RetentionPolicyQueryParams extends QueryParams {
-  private final String durationLiteral;
-
-  protected RetentionPolicyQueryParams(String index, String durationLiteral) {
-    super(index);
-    this.durationLiteral = durationLiteral;
-  }
-
-  public static RetentionPolicyQueryParams from(String index, String durationLiteral) {
-    return new RetentionPolicyQueryParams(index, durationLiteral);
-  }
-
-  public String getDurationLiteral() {
-    return durationLiteral;
-  }
-
-
-}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQuery.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQuery.java
deleted file mode 100644
index 33cd9b997e..0000000000
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQuery.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*
- * 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.
- *
- */
-package org.apache.streampipes.dataexplorer.query;
-
-import org.apache.streampipes.commons.environment.Environment;
-import org.apache.streampipes.commons.environment.Environments;
-import org.apache.streampipes.dataexplorer.commons.influx.InfluxClientProvider;
-import org.apache.streampipes.model.datalake.DataSeries;
-import org.apache.streampipes.model.datalake.SpQueryResult;
-
-import org.influxdb.InfluxDB;
-import org.influxdb.dto.Query;
-
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-
-public abstract class DataExplorerQuery<T> {
-
-  public T executeQuery() throws RuntimeException {
-    InfluxDB influxDB = InfluxClientProvider.getInfluxDBClient();
-    var databaseName = getEnvironment().getTsStorageBucket().getValueOrDefault();
-    DataExplorerQueryBuilder queryBuilder =
-        DataExplorerQueryBuilder.create(databaseName);
-    getQuery(queryBuilder);
-    Query query = queryBuilder.toQuery();
-    org.influxdb.dto.QueryResult result;
-    if (queryBuilder.hasTimeUnit()) {
-      result = influxDB.query(query, queryBuilder.getTimeUnit());
-    } else {
-      result = influxDB.query(query);
-    }
-
-    T dataResult = postQuery(result);
-    influxDB.close();
-
-    return dataResult;
-  }
-
-  protected SpQueryResult convertResult(org.influxdb.dto.QueryResult result) {
-    if (result.getResults().get(0).getSeries() != null) {
-      DataSeries dataSeries = convertResult(result.getResults().get(0).getSeries().get(0));
-      return new SpQueryResult(1, dataSeries.getHeaders(), Arrays.asList(dataSeries));
-    } else {
-      return new SpQueryResult();
-    }
-  }
-
-  protected DataSeries convertResult(org.influxdb.dto.QueryResult.Series serie) {
-    List<String> columns = serie.getColumns();
-    for (int i = 0; i < columns.size(); i++) {
-      String replacedColumnName = columns.get(i).replaceAll("mean_", "");
-      columns.set(i, replacedColumnName);
-    }
-    List values = serie.getValues();
-    return new DataSeries(values.size(), values, columns, new HashMap<>());
-  }
-
-  protected SpQueryResult convertMultiResult(org.influxdb.dto.QueryResult result) {
-    SpQueryResult groupedDataResult = new SpQueryResult();
-    if (result.getResults().get(0).getSeries() != null) {
-      for (org.influxdb.dto.QueryResult.Series series : result.getResults().get(0).getSeries()) {
-        String groupName = series.getTags().entrySet().toArray()[0].toString();
-        DataSeries dataResult = convertResult(series);
-        groupedDataResult.addDataResult(dataResult);
-      }
-    }
-    return groupedDataResult;
-
-  }
-
-  private Environment getEnvironment() {
-    return Environments.getEnvironment();
-  }
-
-  protected abstract void getQuery(DataExplorerQueryBuilder queryBuilder);
-
-  protected abstract T postQuery(org.influxdb.dto.QueryResult result) throws RuntimeException;
-}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQueryBuilder.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQueryBuilder.java
deleted file mode 100644
index 7ca849b68d..0000000000
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQueryBuilder.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * 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.
- *
- */
-package org.apache.streampipes.dataexplorer.query;
-
-import org.influxdb.dto.Query;
-
-import java.util.StringJoiner;
-import java.util.concurrent.TimeUnit;
-
-public class DataExplorerQueryBuilder {
-
-  private StringJoiner queryParts;
-  private String databaseName;
-  private TimeUnit timeUnit;
-
-  private DataExplorerQueryBuilder(String databaseName) {
-    this.queryParts = new StringJoiner(" ");
-    this.databaseName = databaseName;
-  }
-
-  public static DataExplorerQueryBuilder create(String databaseName) {
-    return new DataExplorerQueryBuilder(databaseName);
-  }
-
-  public DataExplorerQueryBuilder add(String queryPart) {
-    this.queryParts.add(queryPart);
-    return this;
-  }
-
-  public DataExplorerQueryBuilder withTimeUnit(TimeUnit timeUnit) {
-    this.timeUnit = timeUnit;
-    return this;
-  }
-
-  public boolean hasTimeUnit() {
-    return this.timeUnit != null;
-  }
-
-  public TimeUnit getTimeUnit() {
-    return this.timeUnit;
-  }
-
-  public Query toQuery() {
-    return new Query(queryParts.toString(), databaseName);
-  }
-}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DeleteDataQuery.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DeleteDataQuery.java
index e4a7a05b9a..6ef1d68564 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DeleteDataQuery.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DeleteDataQuery.java
@@ -17,25 +17,37 @@
  */
 package org.apache.streampipes.dataexplorer.query;
 
+import org.apache.streampipes.commons.environment.Environment;
+import org.apache.streampipes.commons.environment.Environments;
+import org.apache.streampipes.dataexplorer.commons.influx.InfluxClientProvider;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 
+import org.influxdb.InfluxDB;
+import org.influxdb.dto.Query;
 import org.influxdb.dto.QueryResult;
 
-public class DeleteDataQuery extends DataExplorerQuery<QueryResult> {
+public class DeleteDataQuery {
 
-  private DataLakeMeasure measure;
+  private final DataLakeMeasure measure;
 
   public DeleteDataQuery(DataLakeMeasure measure) {
     this.measure = measure;
   }
 
-  @Override
-  protected void getQuery(DataExplorerQueryBuilder queryBuilder) {
-    queryBuilder.add("DROP MEASUREMENT \"" + measure.getMeasureName() + "\"");
+  private String getQuery() {
+    return "DROP MEASUREMENT \"" + measure.getMeasureName() + "\"";
   }
 
-  @Override
-  protected QueryResult postQuery(QueryResult result) {
-    return result;
+  public QueryResult executeQuery() throws RuntimeException {
+    try (final InfluxDB influxDB = InfluxClientProvider.getInfluxDBClient()) {
+      var databaseName = getEnvironment().getTsStorageBucket().getValueOrDefault();
+
+      var query = new Query(getQuery(), databaseName);
+      return influxDB.query(query);
+    }
+  }
+
+  private Environment getEnvironment() {
+    return Environments.getEnvironment();
   }
 }
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/EditRetentionPolicyQuery.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/EditRetentionPolicyQuery.java
deleted file mode 100644
index c28931ad17..0000000000
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/EditRetentionPolicyQuery.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * 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.
- *
- */
-
-package org.apache.streampipes.dataexplorer.query;
-
-import org.apache.streampipes.dataexplorer.param.RetentionPolicyQueryParams;
-
-import org.influxdb.dto.QueryResult;
-
-public class EditRetentionPolicyQuery extends ParameterizedDataExplorerQuery<RetentionPolicyQueryParams, String> {
-
-  private static final String CREATE_OPERATOR = "CREATE";
-  private static final String ALTER_OPERATOR = "ALTER";
-  private static final String DROP_OPERATOR = "DROP";
-  private static final String RESET_OPERATOR = "DEFAULT";
-
-  private String operationToPerform;
-
-  public EditRetentionPolicyQuery(RetentionPolicyQueryParams queryParams, String operation) {
-    super(queryParams);
-    this.operationToPerform = operation;
-  }
-
-
-  @Override
-  protected void getQuery(DataExplorerQueryBuilder queryBuilder) {
-    if (this.operationToPerform.equals(CREATE_OPERATOR)) {
-      queryBuilder.add(createRetentionPolicyStatement(params.getIndex()));
-    } else if (this.operationToPerform.equals(ALTER_OPERATOR)) {
-      queryBuilder.add(alterRetentionPolicyStatement(params.getIndex()));
-    } else if (this.operationToPerform.equals(DROP_OPERATOR)) {
-      queryBuilder.add(dropRetentionPolicyStatement(params.getIndex()));
-    } else if (this.operationToPerform.equals(RESET_OPERATOR)) {
-      queryBuilder.add(resetRetentionPolicyStatement());
-    }
-
-  }
-
-  @Override
-  protected String postQuery(QueryResult result) throws RuntimeException {
-    return result.toString();
-  }
-
-  private String createRetentionPolicyStatement(String index) {
-    return "CREATE RETENTION POLICY " + index + " ON " + "sp DURATION " + params.getDurationLiteral()
-        + " REPLICATION 1 DEFAULT";
-  }
-
-  private String alterRetentionPolicyStatement(String index) {
-    return "ALTER RETENTION POLICY " + index + " ON " + "sp DURATION " + params.getDurationLiteral()
-        + " REPLICATION 1 DEFAULT";
-  }
-
-  private String dropRetentionPolicyStatement(String index) {
-    return "DROP RETENTION POLICY " + index + " ON " + "sp";
-  }
-
-  private String resetRetentionPolicyStatement() {
-    return "ALTER RETENTION POLICY " + "autogen" + " ON " + "sp DURATION " + "0s" + " REPLICATION 1 DEFAULT";
-  }
-
-}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/ParameterizedDataExplorerQuery.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/ParameterizedDataExplorerQuery.java
deleted file mode 100644
index 1c66dfd23c..0000000000
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/ParameterizedDataExplorerQuery.java
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
- * 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.
- *
- */
-package org.apache.streampipes.dataexplorer.query;
-
-public abstract class ParameterizedDataExplorerQuery<K, V> extends DataExplorerQuery<V> {
-
-  protected K params;
-
-  public ParameterizedDataExplorerQuery(K queryParams) {
-    this.params = queryParams;
-  }
-
-}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/ShowRetentionPolicyQuery.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/ShowRetentionPolicyQuery.java
deleted file mode 100644
index 4ccbb19f43..0000000000
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/ShowRetentionPolicyQuery.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * 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.
- *
- */
-
-package org.apache.streampipes.dataexplorer.query;
-
-import org.apache.streampipes.dataexplorer.param.RetentionPolicyQueryParams;
-import org.apache.streampipes.model.datalake.DataLakeRetentionPolicy;
-
-import org.influxdb.dto.QueryResult;
-
-import java.util.ArrayList;
-import java.util.List;
-
-public class ShowRetentionPolicyQuery
-    extends ParameterizedDataExplorerQuery<RetentionPolicyQueryParams, List<DataLakeRetentionPolicy>> {
-
-  public ShowRetentionPolicyQuery(RetentionPolicyQueryParams queryParams) {
-    super(queryParams);
-  }
-
-
-  @Override
-  protected void getQuery(DataExplorerQueryBuilder queryBuilder) {
-    queryBuilder.add(showRetentionPolicyStatement());
-  }
-
-  @Override
-  protected List<DataLakeRetentionPolicy> postQuery(QueryResult result) throws RuntimeException {
-    List<DataLakeRetentionPolicy> policies = new ArrayList<>();
-    for (List<Object> a : result.getResults().get(0).getSeries().get(0).getValues()) {
-      boolean isDefault = false;
-      if (a.get(4).toString().equals("true")) {
-        isDefault = true;
-      }
-      policies.add(new DataLakeRetentionPolicy(a.get(0).toString(), a.get(1).toString(), isDefault));
-    }
-    return policies;
-  }
-
-  private String showRetentionPolicyStatement() {
-    return "SHOW RETENTION POLICIES  ON " + "sp";
-  }
-}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/template/QueryTemplates.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/template/QueryTemplates.java
deleted file mode 100644
index 80ff5d1fa6..0000000000
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/template/QueryTemplates.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * 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.
- *
- */
-package org.apache.streampipes.dataexplorer.template;
-
-public class QueryTemplates {
-
-  public static String selectWildcardFrom(String index) {
-    return "SELECT * FROM " + index;
-  }
-
-  public static String selectMeanFrom(String index) {
-    return "SELECT mean(*) FROM " + index;
-  }
-
-  public static String selectCountFrom(String index) {
-    return "SELECT count(*) FROM " + index;
-  }
-
-  public static String whereTimeWithin(long startDate, long endDate) {
-    return "WHERE time > "
-        + startDate * 1000000
-        + " AND time < "
-        + endDate * 1000000;
-  }
-}
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/AutoAggregationHandler.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/AutoAggregationHandler.java
index 63d9bd9fae..a63c980778 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/AutoAggregationHandler.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/AutoAggregationHandler.java
@@ -18,7 +18,7 @@
 package org.apache.streampipes.dataexplorer.v4;
 
 import org.apache.streampipes.dataexplorer.DataLakeManagementV4;
-import org.apache.streampipes.dataexplorer.model.Order;
+import org.apache.streampipes.dataexplorer.sdk.DataLakeQueryOrdering;
 import org.apache.streampipes.dataexplorer.v4.params.SelectColumn;
 import org.apache.streampipes.model.datalake.SpQueryResult;
 
@@ -59,10 +59,9 @@ public class AutoAggregationHandler {
   }
 
   public ProvidedQueryParams makeAutoAggregationQueryParams() throws IllegalArgumentException {
-    //checkAllArgumentsPresent();
     try {
-      SpQueryResult newest = getSingleRecord(Order.DESC);
-      SpQueryResult oldest = getSingleRecord(Order.ASC);
+      SpQueryResult newest = getSingleRecord(DataLakeQueryOrdering.DESC);
+      SpQueryResult oldest = getSingleRecord(DataLakeQueryOrdering.ASC);
       if (newest.getTotal() > 0) {
         String sampleField = getSampleField(newest);
         Integer count = getCount(sampleField);
@@ -86,13 +85,6 @@ public class AutoAggregationHandler {
     return null;
   }
 
-  private void checkAllArgumentsPresent() throws IllegalArgumentException {
-    if (!this.queryParams.has(QP_AGGREGATION_FUNCTION)) {
-      throw new IllegalArgumentException(
-          "Auto-Aggregate must provide one of the aggregationFunction parameters MEAN, FIRST, LAST.");
-    }
-  }
-
   private ProvidedQueryParams disableAutoAgg(ProvidedQueryParams params) {
     params.remove(QP_AUTO_AGGREGATE);
     return params;
@@ -120,11 +112,11 @@ public class AutoAggregationHandler {
     return Double.valueOf(v).intValue();
   }
 
-  private SpQueryResult getSingleRecord(Order order) throws ParseException {
+  private SpQueryResult getSingleRecord(DataLakeQueryOrdering order) throws ParseException {
     ProvidedQueryParams singleEvent = disableAutoAgg(new ProvidedQueryParams(queryParams));
     singleEvent.remove(QP_AGGREGATION_FUNCTION);
     singleEvent.update(QP_LIMIT, 1);
-    singleEvent.update(QP_ORDER, order.toValue());
+    singleEvent.update(QP_ORDER, order.name());
     singleEvent.update(QP_COLUMNS, transformColumns(singleEvent.getAsString(QP_COLUMNS)));
 
     return fireQuery(singleEvent);
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/template/QueryTemplatesV4.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/template/QueryTemplatesV4.java
index 9b0392cdb7..5ada24d774 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/template/QueryTemplatesV4.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/v4/template/QueryTemplatesV4.java
@@ -17,34 +17,8 @@
  */
 package org.apache.streampipes.dataexplorer.v4.template;
 
-import java.util.StringJoiner;
-
 public class QueryTemplatesV4 {
 
-  public static String selectFrom(String index, String columns) {
-    return "SELECT " + columns + " FROM " + index;
-  }
-
-  public static String selectCountFrom(String index, String selectedColumns) {
-    return selectAggregationFrom(index, selectedColumns, "COUNT");
-  }
-
-  public static String selectAggregationFrom(String index, String columns, String aggregationFunction) {
-    String[] cols = columns.split(",");
-    StringJoiner joiner = new StringJoiner(", ");
-    //StringBuilder statement = new StringBuilder(aggregationFunction + "(" + cols[0] + ")");
-
-    for (int i = 0; i < cols.length; i++) {
-      String builder = aggregationFunction
-          + "("
-          + cols[i]
-          + ")" + " AS " + cols[i];
-      joiner.add(builder);
-    }
-
-    return "SELECT " + joiner + " FROM \"" + index + "\"";
-  }
-
   public static String deleteFrom(String index) {
     return "DELETE FROM \"" + index + "\"";
   }
diff --git a/streampipes-extensions/streampipes-processors-transformation-jvm/pom.xml b/streampipes-extensions/streampipes-processors-transformation-jvm/pom.xml
index 49f17ed967..8bc09ed41c 100644
--- a/streampipes-extensions/streampipes-processors-transformation-jvm/pom.xml
+++ b/streampipes-extensions/streampipes-processors-transformation-jvm/pom.xml
@@ -47,6 +47,12 @@
             <artifactId>streampipes-wrapper-standalone</artifactId>
             <version>0.92.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-test-utils</artifactId>
+            <version>0.92.0-SNAPSHOT</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
diff --git a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/TransformationJvmInit.java b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/TransformationJvmInit.java
index c669c67ba5..b060d08583 100644
--- a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/TransformationJvmInit.java
+++ b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/TransformationJvmInit.java
@@ -29,9 +29,9 @@ import org.apache.streampipes.messaging.kafka.SpKafkaProtocolFactory;
 import org.apache.streampipes.messaging.mqtt.SpMqttProtocolFactory;
 import org.apache.streampipes.processors.transformation.jvm.processor.array.count.CountArrayController;
 import org.apache.streampipes.processors.transformation.jvm.processor.array.split.SplitArrayController;
-import org.apache.streampipes.processors.transformation.jvm.processor.booloperator.counter.BooleanCounterController;
+import org.apache.streampipes.processors.transformation.jvm.processor.booloperator.counter.BooleanCounterProcessor;
 import org.apache.streampipes.processors.transformation.jvm.processor.booloperator.edge.SignalEdgeFilterController;
-import org.apache.streampipes.processors.transformation.jvm.processor.booloperator.inverter.BooleanInverterController;
+import org.apache.streampipes.processors.transformation.jvm.processor.booloperator.inverter.BooleanInverterProcessor;
 import org.apache.streampipes.processors.transformation.jvm.processor.booloperator.logical.BooleanOperatorProcessor;
 import org.apache.streampipes.processors.transformation.jvm.processor.booloperator.state.BooleanToStateController;
 import org.apache.streampipes.processors.transformation.jvm.processor.booloperator.timekeeping.BooleanTimekeepingController;
@@ -70,8 +70,8 @@ public class TransformationJvmInit extends ExtensionsModelSubmitter {
             new CalculateDurationController(),
             new ChangedValueDetectionController(),
             new TimestampExtractorController(),
-            new BooleanCounterController(),
-            new BooleanInverterController(),
+            new BooleanCounterProcessor(),
+            new BooleanInverterProcessor(),
             new BooleanTimekeepingController(),
             new BooleanTimerController(),
             new CsvMetadataEnrichmentController(),
diff --git a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/counter/BooleanCounter.java b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/counter/BooleanCounter.java
deleted file mode 100644
index 5aeb279ac2..0000000000
--- a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/counter/BooleanCounter.java
+++ /dev/null
@@ -1,92 +0,0 @@
-/*
- * 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.
- *
- */
-
-package org.apache.streampipes.processors.transformation.jvm.processor.booloperator.counter;
-
-import org.apache.streampipes.logging.api.Logger;
-import org.apache.streampipes.model.runtime.Event;
-import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext;
-import org.apache.streampipes.wrapper.routing.SpOutputCollector;
-import org.apache.streampipes.wrapper.runtime.EventProcessor;
-
-public class BooleanCounter implements EventProcessor<BooleanCounterParameters> {
-
-  // From true to false or from false to true
-
-  private static Logger log;
-
-  private String fieldName;
-  private int flankUp;
-
-  private boolean fieldValueOfLastEvent;
-  private int counter;
-
-
-  @Override
-  public void onInvocation(BooleanCounterParameters booleanCounterParametersParameters,
-                           SpOutputCollector spOutputCollector,
-                           EventProcessorRuntimeContext runtimeContext) {
-    log = booleanCounterParametersParameters.getGraph().getLogger(BooleanCounter.class);
-    this.fieldName = booleanCounterParametersParameters.getInvertFieldName();
-    this.flankUp = booleanCounterParametersParameters.getFlankUp();
-
-    if (flankUp == 1) {
-      this.fieldValueOfLastEvent = true;
-    } else {
-      this.fieldValueOfLastEvent = false;
-    }
-
-    this.counter = 0;
-  }
-
-  @Override
-  public void onEvent(Event inputEvent, SpOutputCollector out) {
-
-    boolean value = inputEvent.getFieldBySelector(fieldName).getAsPrimitive().getAsBoolean();
-    boolean updateCounter = false;
-
-    if (this.flankUp == 2) {
-      // detect up flanks
-      if (!this.fieldValueOfLastEvent && value) {
-        updateCounter = true;
-      }
-    } else if (this.flankUp == 1) {
-      // detect up flanks
-      if (this.fieldValueOfLastEvent && !value) {
-        updateCounter = true;
-      }
-    } else {
-      if (this.fieldValueOfLastEvent != value) {
-        updateCounter = true;
-      }
-    }
-
-    if (updateCounter) {
-      this.counter++;
-      inputEvent.addField(BooleanCounterController.COUNT_FIELD_RUNTIME_NAME, this.counter);
-      out.collect(inputEvent);
-    }
-
-    this.fieldValueOfLastEvent = value;
-  }
-
-  @Override
-  public void onDetach() {
-  }
-
-}
diff --git a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/counter/BooleanCounterController.java b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/counter/BooleanCounterController.java
deleted file mode 100644
index 69e1cc2a39..0000000000
--- a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/counter/BooleanCounterController.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * 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.
- *
- */
-
-package org.apache.streampipes.processors.transformation.jvm.processor.booloperator.counter;
-
-import org.apache.streampipes.model.DataProcessorType;
-import org.apache.streampipes.model.graph.DataProcessorDescription;
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import org.apache.streampipes.model.schema.PropertyScope;
-import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
-import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
-import org.apache.streampipes.sdk.extractor.ProcessingElementParameterExtractor;
-import org.apache.streampipes.sdk.helpers.EpProperties;
-import org.apache.streampipes.sdk.helpers.EpRequirements;
-import org.apache.streampipes.sdk.helpers.Labels;
-import org.apache.streampipes.sdk.helpers.Locales;
-import org.apache.streampipes.sdk.helpers.Options;
-import org.apache.streampipes.sdk.helpers.OutputStrategies;
-import org.apache.streampipes.sdk.utils.Assets;
-import org.apache.streampipes.wrapper.standalone.ConfiguredEventProcessor;
-import org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
-
-public class BooleanCounterController extends StandaloneEventProcessingDeclarer<BooleanCounterParameters> {
-
-  public static final String FIELD_ID = "field";
-  public static final String FLANK_ID = "flank";
-  public static final String COUNT_FIELD_ID = "countField";
-
-  public static final String COUNT_FIELD_RUNTIME_NAME = "counter";
-
-  private static final String FLANK_UP = "FALSE -> TRUE";
-  private static final String FLANK_DOWN = "TRUE -> FALSE";
-  private static final String BOTH = "BOTH";
-
-  @Override
-  public DataProcessorDescription declareModel() {
-    return ProcessingElementBuilder.create("org.apache.streampipes.processors.transformation.jvm.booloperator.counter")
-        .category(DataProcessorType.BOOLEAN_OPERATOR, DataProcessorType.COUNT_OPERATOR)
-        .withLocales(Locales.EN)
-        .withAssets(Assets.DOCUMENTATION, Assets.ICON)
-        .requiredStream(StreamRequirementsBuilder.create()
-            .requiredPropertyWithUnaryMapping(
-                EpRequirements.booleanReq(),
-                Labels.withId(FIELD_ID),
-                PropertyScope.NONE)
-            .build())
-
-        .requiredSingleValueSelection(Labels.withId(FLANK_ID), Options.from(BOTH, FLANK_UP, FLANK_DOWN))
-        .outputStrategy(OutputStrategies.append(
-            EpProperties.numberEp(Labels.withId(COUNT_FIELD_ID), COUNT_FIELD_RUNTIME_NAME, "http://schema.org/Number")
-        ))
-        .build();
-  }
-
-  @Override
-  public ConfiguredEventProcessor<BooleanCounterParameters> onInvocation(
-      DataProcessorInvocation graph,
-      ProcessingElementParameterExtractor extractor) {
-
-    String invertFieldName = extractor.mappingPropertyValue(FIELD_ID);
-    String flank = extractor.selectedSingleValue(FLANK_ID, String.class);
-
-    int flankUp = 0;
-    if (flank.equals(FLANK_DOWN)) {
-      flankUp = 1;
-    } else if (flank.equals(FLANK_UP)) {
-      flankUp = 2;
-    }
-
-    BooleanCounterParameters params = new BooleanCounterParameters(graph, invertFieldName, flankUp);
-    return new ConfiguredEventProcessor<>(params, BooleanCounter::new);
-  }
-}
diff --git a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/counter/BooleanCounterParameters.java b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/counter/BooleanCounterParameters.java
deleted file mode 100644
index 5971d8a70c..0000000000
--- a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/counter/BooleanCounterParameters.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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.
- *
- */
-
-package org.apache.streampipes.processors.transformation.jvm.processor.booloperator.counter;
-
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams;
-
-public class BooleanCounterParameters extends EventProcessorBindingParams {
-  private String invertFieldName;
-
-  /**
-   * Defines which boolean changes should be counted
-   * 0: BOTH
-   * 1: TRUE -> FALSE
-   * 2: FALSE -> TRUE
-   */
-  private int flankUp;
-
-  public BooleanCounterParameters(DataProcessorInvocation graph, String invertFieldName, int flankUp) {
-    super(graph);
-    this.invertFieldName = invertFieldName;
-    this.flankUp = flankUp;
-  }
-
-  public String getInvertFieldName() {
-    return invertFieldName;
-  }
-
-  public int getFlankUp() {
-    return flankUp;
-  }
-}
diff --git a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/counter/BooleanCounterProcessor.java b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/counter/BooleanCounterProcessor.java
new file mode 100644
index 0000000000..5de9066430
--- /dev/null
+++ b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/counter/BooleanCounterProcessor.java
@@ -0,0 +1,140 @@
+/*
+ * 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.
+ *
+ */
+
+package org.apache.streampipes.processors.transformation.jvm.processor.booloperator.counter;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.logging.api.Logger;
+import org.apache.streampipes.model.DataProcessorType;
+import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.schema.PropertyScope;
+import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
+import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+import org.apache.streampipes.sdk.extractor.ProcessingElementParameterExtractor;
+import org.apache.streampipes.sdk.helpers.EpProperties;
+import org.apache.streampipes.sdk.helpers.EpRequirements;
+import org.apache.streampipes.sdk.helpers.Labels;
+import org.apache.streampipes.sdk.helpers.Locales;
+import org.apache.streampipes.sdk.helpers.Options;
+import org.apache.streampipes.sdk.helpers.OutputStrategies;
+import org.apache.streampipes.sdk.utils.Assets;
+import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext;
+import org.apache.streampipes.wrapper.routing.SpOutputCollector;
+import org.apache.streampipes.wrapper.standalone.ProcessorParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataProcessor;
+
+public class BooleanCounterProcessor extends StreamPipesDataProcessor {
+  public static final String FIELD_ID = "field";
+  public static final String FLANK_ID = "flank";
+  public static final String COUNT_FIELD_ID = "countField";
+  public static final String COUNT_FIELD_RUNTIME_NAME = "counter";
+  private static final String FLANK_UP = "FALSE -> TRUE";
+  private static final String FLANK_DOWN = "TRUE -> FALSE";
+  private static final String BOTH = "BOTH";
+  private static Logger log;
+  private String fieldName;
+  /**
+   * Defines which boolean changes should be counted
+   * 0: BOTH
+   * 1: TRUE -> FALSE
+   * 2: FALSE -> TRUE
+   */
+  private int flankUp;
+  private boolean fieldValueOfLastEvent;
+  private int counter;
+  @Override
+  public DataProcessorDescription declareModel() {
+    return ProcessingElementBuilder.create("org.apache.streampipes.processors.transformation.jvm.booloperator.counter")
+            .category(DataProcessorType.BOOLEAN_OPERATOR, DataProcessorType.COUNT_OPERATOR)
+            .withLocales(Locales.EN)
+            .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+            .requiredStream(StreamRequirementsBuilder.create()
+                    .requiredPropertyWithUnaryMapping(
+                            EpRequirements.booleanReq(),
+                            Labels.withId(FIELD_ID),
+                            PropertyScope.NONE)
+                    .build())
+
+            .requiredSingleValueSelection(Labels.withId(FLANK_ID), Options.from(BOTH, FLANK_UP, FLANK_DOWN))
+            .outputStrategy(OutputStrategies
+                    .append(EpProperties
+                            .numberEp(
+                                    Labels.withId(COUNT_FIELD_ID),
+                                      COUNT_FIELD_RUNTIME_NAME,
+                                      "http://schema.org/Number")
+              ))
+              .build();
+  }
+
+  @Override
+  public void onInvocation(ProcessorParams parameters,
+                           SpOutputCollector spOutputCollector,
+                           EventProcessorRuntimeContext runtimeContext) throws SpRuntimeException {
+    ProcessingElementParameterExtractor extractor = parameters.extractor();
+    log = parameters.getGraph().getLogger(BooleanCounterProcessor.class);
+    String flank = extractor.selectedSingleValue(FLANK_ID, String.class);
+    this.fieldName = extractor.mappingPropertyValue(FIELD_ID);
+    this.flankUp = 0;
+
+    if (flank.equals(FLANK_DOWN)) {
+      this.flankUp = 1;
+      this.fieldValueOfLastEvent = true;
+    } else if (flank.equals(FLANK_UP)) {
+      this.flankUp = 2;
+      this.fieldValueOfLastEvent = false;
+    }
+
+    this.counter = 0;
+  }
+
+  @Override
+  public void onEvent(Event inputEvent, SpOutputCollector out) throws SpRuntimeException {
+
+    boolean value = inputEvent.getFieldBySelector(fieldName).getAsPrimitive().getAsBoolean();
+    boolean updateCounter = false;
+
+    if (this.flankUp == 2) {
+      // detect up flanks
+      if (!this.fieldValueOfLastEvent && value) {
+        updateCounter = true;
+      }
+    } else if (this.flankUp == 1) {
+      // detect up flanks
+      if (this.fieldValueOfLastEvent && !value) {
+        updateCounter = true;
+      }
+    } else {
+      if (this.fieldValueOfLastEvent != value) {
+        updateCounter = true;
+      }
+    }
+
+    if (updateCounter) {
+      this.counter++;
+      inputEvent.addField(COUNT_FIELD_RUNTIME_NAME, this.counter);
+      out.collect(inputEvent);
+    }
+
+    this.fieldValueOfLastEvent = value;
+  }
+
+  @Override
+  public void onDetach() {
+  }
+}
\ No newline at end of file
diff --git a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/inverter/BooleanInverter.java b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/inverter/BooleanInverter.java
deleted file mode 100644
index a5137d5ec6..0000000000
--- a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/inverter/BooleanInverter.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * 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.
- *
- */
-
-package org.apache.streampipes.processors.transformation.jvm.processor.booloperator.inverter;
-
-import org.apache.streampipes.logging.api.Logger;
-import org.apache.streampipes.model.runtime.Event;
-import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext;
-import org.apache.streampipes.wrapper.routing.SpOutputCollector;
-import org.apache.streampipes.wrapper.runtime.EventProcessor;
-
-public class BooleanInverter implements EventProcessor<BooleanInverterParameters> {
-
-  private static Logger log;
-
-  private String invertFieldName;
-
-
-  @Override
-  public void onInvocation(BooleanInverterParameters booleanInverterParameters,
-                           SpOutputCollector spOutputCollector,
-                           EventProcessorRuntimeContext runtimeContext) {
-    log = booleanInverterParameters.getGraph().getLogger(BooleanInverter.class);
-    this.invertFieldName = booleanInverterParameters.getInvertFieldName();
-  }
-
-  @Override
-  public void onEvent(Event inputEvent, SpOutputCollector out) {
-
-    boolean field = inputEvent.getFieldBySelector(invertFieldName).getAsPrimitive().getAsBoolean();
-    inputEvent.updateFieldBySelector(invertFieldName, !field);
-
-    out.collect(inputEvent);
-  }
-
-  @Override
-  public void onDetach() {
-  }
-}
diff --git a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/inverter/BooleanInverterParameters.java b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/inverter/BooleanInverterParameters.java
deleted file mode 100644
index 636bb79085..0000000000
--- a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/inverter/BooleanInverterParameters.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * 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.
- *
- */
-
-package org.apache.streampipes.processors.transformation.jvm.processor.booloperator.inverter;
-
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams;
-
-public class BooleanInverterParameters extends EventProcessorBindingParams {
-  private String invertFieldName;
-
-  public BooleanInverterParameters(DataProcessorInvocation graph, String invertFieldName) {
-    super(graph);
-    this.invertFieldName = invertFieldName;
-  }
-
-  public String getInvertFieldName() {
-    return invertFieldName;
-  }
-}
diff --git a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/inverter/BooleanInverterController.java b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/inverter/BooleanInverterProcessor.java
similarity index 62%
rename from streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/inverter/BooleanInverterController.java
rename to streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/inverter/BooleanInverterProcessor.java
index 8ae674e07e..7c75ed72b3 100644
--- a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/inverter/BooleanInverterController.java
+++ b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/inverter/BooleanInverterProcessor.java
@@ -18,9 +18,11 @@
 
 package org.apache.streampipes.processors.transformation.jvm.processor.booloperator.inverter;
 
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.logging.api.Logger;
 import org.apache.streampipes.model.DataProcessorType;
 import org.apache.streampipes.model.graph.DataProcessorDescription;
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.runtime.Event;
 import org.apache.streampipes.model.schema.PropertyScope;
 import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
 import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
@@ -30,13 +32,15 @@ import org.apache.streampipes.sdk.helpers.Labels;
 import org.apache.streampipes.sdk.helpers.Locales;
 import org.apache.streampipes.sdk.helpers.OutputStrategies;
 import org.apache.streampipes.sdk.utils.Assets;
-import org.apache.streampipes.wrapper.standalone.ConfiguredEventProcessor;
-import org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
-
-public class BooleanInverterController extends StandaloneEventProcessingDeclarer<BooleanInverterParameters> {
+import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext;
+import org.apache.streampipes.wrapper.routing.SpOutputCollector;
+import org.apache.streampipes.wrapper.standalone.ProcessorParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataProcessor;
 
+public class BooleanInverterProcessor extends StreamPipesDataProcessor {
   public static final String INVERT_FIELD_ID = "invert-field";
-
+  private static Logger log;
+  private String invertFieldName;
   @Override
   public DataProcessorDescription declareModel() {
     return ProcessingElementBuilder.create("org.apache.streampipes.processors.transformation.jvm.booloperator.inverter")
@@ -54,13 +58,23 @@ public class BooleanInverterController extends StandaloneEventProcessingDeclarer
   }
 
   @Override
-  public ConfiguredEventProcessor<BooleanInverterParameters> onInvocation(
-      DataProcessorInvocation graph,
-      ProcessingElementParameterExtractor extractor) {
+  public void onInvocation(ProcessorParams parameters,
+                           SpOutputCollector spOutputCollector,
+                           EventProcessorRuntimeContext runtimeContext) throws SpRuntimeException {
+    ProcessingElementParameterExtractor extractor = parameters.extractor();
+    log = parameters.getGraph().getLogger(BooleanInverterProcessor.class);
+    this.invertFieldName = extractor.mappingPropertyValue(INVERT_FIELD_ID);
+  }
 
-    String invertFieldName = extractor.mappingPropertyValue(INVERT_FIELD_ID);
-    BooleanInverterParameters params = new BooleanInverterParameters(graph, invertFieldName);
+  @Override
+  public void onEvent(Event inputEvent, SpOutputCollector collector) throws SpRuntimeException {
+    boolean field = inputEvent.getFieldBySelector(invertFieldName).getAsPrimitive().getAsBoolean();
+    inputEvent.updateFieldBySelector(invertFieldName, !field);
+    collector.collect(inputEvent);
+  }
+
+  @Override
+  public void onDetach() throws SpRuntimeException {
 
-    return new ConfiguredEventProcessor<>(params, BooleanInverter::new);
   }
-}
+}
\ No newline at end of file
diff --git a/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeResourceV4.java b/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeResourceV4.java
index 9186564baf..4ed1a0bd78 100644
--- a/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeResourceV4.java
+++ b/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeResourceV4.java
@@ -22,7 +22,6 @@ import org.apache.streampipes.dataexplorer.DataLakeManagementV4;
 import org.apache.streampipes.dataexplorer.v4.ProvidedQueryParams;
 import org.apache.streampipes.dataexplorer.v4.query.writer.OutputFormat;
 import org.apache.streampipes.model.StreamPipesErrorMessage;
-import org.apache.streampipes.model.datalake.DataLakeConfiguration;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.model.datalake.DataSeries;
 import org.apache.streampipes.model.datalake.SpQueryResult;
@@ -40,7 +39,6 @@ import org.slf4j.LoggerFactory;
 
 import jakarta.ws.rs.Consumes;
 import jakarta.ws.rs.DELETE;
-import jakarta.ws.rs.DefaultValue;
 import jakarta.ws.rs.GET;
 import jakarta.ws.rs.POST;
 import jakarta.ws.rs.Path;
@@ -94,18 +92,6 @@ public class DataLakeResourceV4 extends AbstractRestResource {
   }
 
 
-  @POST
-  @Path("/configuration")
-  @Consumes(MediaType.APPLICATION_JSON)
-  @Operation(summary = "Configure the parameters of the data lake", tags = {"Data Lake"},
-      responses = {@ApiResponse(responseCode = "200", description = "Configuration was successful")})
-  public Response configureMeasurement(
-      @Parameter(in = ParameterIn.QUERY, description = "should any parameter be reset to its default value?")
-      @DefaultValue("false") @QueryParam("resetToDefault") boolean resetToDefault,
-      @Parameter(in = ParameterIn.DEFAULT, description = "the configuration parameters") DataLakeConfiguration config) {
-    return ok(this.dataLakeManagement.editMeasurementConfiguration(config, resetToDefault));
-  }
-
   @DELETE
   @Path("/measurements/{measurementID}")
   @Operation(summary = "Remove data from a single measurement series with given id", tags = {"Data Lake"},
@@ -349,22 +335,6 @@ public class DataLakeResourceV4 extends AbstractRestResource {
     }
   }
 
-
-  @GET
-  @Path("/configuration")
-  @Produces(MediaType.APPLICATION_JSON)
-  @Operation(summary = "Get the configuration parameters of the data lake", tags = {"Data Lake"},
-      responses = {
-          @ApiResponse(
-              responseCode = "200",
-              description = "configuration parameters",
-              content = @Content(schema = @Schema(implementation = DataLakeConfiguration.class)))})
-  public Response getMeasurementConfiguration(
-      @Parameter(in = ParameterIn.QUERY, description = "the id of a specific configuration parameter")
-      @QueryParam("parameterID") String parameterID) {
-    return ok(this.dataLakeManagement.getDataLakeConfiguration());
-  }
-
   @DELETE
   @Path("/measurements")
   @Operation(summary = "Remove all stored measurement series from Data Lake", tags = {"Data Lake"},
diff --git a/ui/package-lock.json b/ui/package-lock.json
index 15a39155ff..c34d70b6d3 100644
--- a/ui/package-lock.json
+++ b/ui/package-lock.json
@@ -92,7 +92,7 @@
         "@types/rx": "^4.1.2",
         "@types/showdown": "1.9.4",
         "@typescript-eslint/eslint-plugin": "5.53.0",
-        "@typescript-eslint/parser": "5.52.0",
+        "@typescript-eslint/parser": "5.54.0",
         "ajv-keywords": "^5.1.0",
         "assert": "^2.0.0",
         "compression-webpack-plugin": "^10.0.0",
@@ -6331,14 +6331,14 @@
       }
     },
     "node_modules/@typescript-eslint/parser": {
-      "version": "5.52.0",
-      "resolved": "https://registry.npmjs.org/@typescript-eslint/parser/-/parser-5.52.0.tgz",
-      "integrity": "sha512-e2KiLQOZRo4Y0D/b+3y08i3jsekoSkOYStROYmPUnGMEoA0h+k2qOH5H6tcjIc68WDvGwH+PaOrP1XRzLJ6QlA==",
+      "version": "5.54.0",
+      "resolved": "https://registry.npmjs.org/@typescript-eslint/parser/-/parser-5.54.0.tgz",
+      "integrity": "sha512-aAVL3Mu2qTi+h/r04WI/5PfNWvO6pdhpeMRWk9R7rEV4mwJNzoWf5CCU5vDKBsPIFQFjEq1xg7XBI2rjiMXQbQ==",
       "dev": true,
       "dependencies": {
-        "@typescript-eslint/scope-manager": "5.52.0",
-        "@typescript-eslint/types": "5.52.0",
-        "@typescript-eslint/typescript-estree": "5.52.0",
+        "@typescript-eslint/scope-manager": "5.54.0",
+        "@typescript-eslint/types": "5.54.0",
+        "@typescript-eslint/typescript-estree": "5.54.0",
         "debug": "^4.3.4"
       },
       "engines": {
@@ -6358,13 +6358,13 @@
       }
     },
     "node_modules/@typescript-eslint/scope-manager": {
-      "version": "5.52.0",
-      "resolved": "https://registry.npmjs.org/@typescript-eslint/scope-manager/-/scope-manager-5.52.0.tgz",
-      "integrity": "sha512-AR7sxxfBKiNV0FWBSARxM8DmNxrwgnYMPwmpkC1Pl1n+eT8/I2NAUPuwDy/FmDcC6F8pBfmOcaxcxRHspgOBMw==",
+      "version": "5.54.0",
+      "resolved": "https://registry.npmjs.org/@typescript-eslint/scope-manager/-/scope-manager-5.54.0.tgz",
+      "integrity": "sha512-VTPYNZ7vaWtYna9M4oD42zENOBrb+ZYyCNdFs949GcN8Miwn37b8b7eMj+EZaq7VK9fx0Jd+JhmkhjFhvnovhg==",
       "dev": true,
       "dependencies": {
-        "@typescript-eslint/types": "5.52.0",
-        "@typescript-eslint/visitor-keys": "5.52.0"
+        "@typescript-eslint/types": "5.54.0",
+        "@typescript-eslint/visitor-keys": "5.54.0"
       },
       "engines": {
         "node": "^12.22.0 || ^14.17.0 || >=16.0.0"
@@ -6524,9 +6524,9 @@
       }
     },
     "node_modules/@typescript-eslint/types": {
-      "version": "5.52.0",
-      "resolved": "https://registry.npmjs.org/@typescript-eslint/types/-/types-5.52.0.tgz",
-      "integrity": "sha512-oV7XU4CHYfBhk78fS7tkum+/Dpgsfi91IIDy7fjCyq2k6KB63M6gMC0YIvy+iABzmXThCRI6xpCEyVObBdWSDQ==",
+      "version": "5.54.0",
+      "resolved": "https://registry.npmjs.org/@typescript-eslint/types/-/types-5.54.0.tgz",
+      "integrity": "sha512-nExy+fDCBEgqblasfeE3aQ3NuafBUxZxgxXcYfzYRZFHdVvk5q60KhCSkG0noHgHRo/xQ/BOzURLZAafFpTkmQ==",
       "dev": true,
       "engines": {
         "node": "^12.22.0 || ^14.17.0 || >=16.0.0"
@@ -6537,13 +6537,13 @@
       }
     },
     "node_modules/@typescript-eslint/typescript-estree": {
-      "version": "5.52.0",
-      "resolved": "https://registry.npmjs.org/@typescript-eslint/typescript-estree/-/typescript-estree-5.52.0.tgz",
-      "integrity": "sha512-WeWnjanyEwt6+fVrSR0MYgEpUAuROxuAH516WPjUblIrClzYJj0kBbjdnbQXLpgAN8qbEuGywiQsXUVDiAoEuQ==",
+      "version": "5.54.0",
+      "resolved": "https://registry.npmjs.org/@typescript-eslint/typescript-estree/-/typescript-estree-5.54.0.tgz",
+      "integrity": "sha512-X2rJG97Wj/VRo5YxJ8Qx26Zqf0RRKsVHd4sav8NElhbZzhpBI8jU54i6hfo9eheumj4oO4dcRN1B/zIVEqR/MQ==",
       "dev": true,
       "dependencies": {
-        "@typescript-eslint/types": "5.52.0",
-        "@typescript-eslint/visitor-keys": "5.52.0",
+        "@typescript-eslint/types": "5.54.0",
+        "@typescript-eslint/visitor-keys": "5.54.0",
         "debug": "^4.3.4",
         "globby": "^11.1.0",
         "is-glob": "^4.0.3",
@@ -6686,12 +6686,12 @@
       }
     },
     "node_modules/@typescript-eslint/visitor-keys": {
-      "version": "5.52.0",
-      "resolved": "https://registry.npmjs.org/@typescript-eslint/visitor-keys/-/visitor-keys-5.52.0.tgz",
-      "integrity": "sha512-qMwpw6SU5VHCPr99y274xhbm+PRViK/NATY6qzt+Et7+mThGuFSl/ompj2/hrBlRP/kq+BFdgagnOSgw9TB0eA==",
+      "version": "5.54.0",
+      "resolved": "https://registry.npmjs.org/@typescript-eslint/visitor-keys/-/visitor-keys-5.54.0.tgz",
+      "integrity": "sha512-xu4wT7aRCakGINTLGeyGqDn+78BwFlggwBjnHa1ar/KaGagnmwLYmlrXIrgAaQ3AE1Vd6nLfKASm7LrFHNbKGA==",
       "dev": true,
       "dependencies": {
-        "@typescript-eslint/types": "5.52.0",
+        "@typescript-eslint/types": "5.54.0",
         "eslint-visitor-keys": "^3.3.0"
       },
       "engines": {
@@ -31144,25 +31144,25 @@
       }
     },
     "@typescript-eslint/parser": {
-      "version": "5.52.0",
-      "resolved": "https://registry.npmjs.org/@typescript-eslint/parser/-/parser-5.52.0.tgz",
-      "integrity": "sha512-e2KiLQOZRo4Y0D/b+3y08i3jsekoSkOYStROYmPUnGMEoA0h+k2qOH5H6tcjIc68WDvGwH+PaOrP1XRzLJ6QlA==",
+      "version": "5.54.0",
+      "resolved": "https://registry.npmjs.org/@typescript-eslint/parser/-/parser-5.54.0.tgz",
+      "integrity": "sha512-aAVL3Mu2qTi+h/r04WI/5PfNWvO6pdhpeMRWk9R7rEV4mwJNzoWf5CCU5vDKBsPIFQFjEq1xg7XBI2rjiMXQbQ==",
       "dev": true,
       "requires": {
-        "@typescript-eslint/scope-manager": "5.52.0",
-        "@typescript-eslint/types": "5.52.0",
-        "@typescript-eslint/typescript-estree": "5.52.0",
+        "@typescript-eslint/scope-manager": "5.54.0",
+        "@typescript-eslint/types": "5.54.0",
+        "@typescript-eslint/typescript-estree": "5.54.0",
         "debug": "^4.3.4"
       }
     },
     "@typescript-eslint/scope-manager": {
-      "version": "5.52.0",
-      "resolved": "https://registry.npmjs.org/@typescript-eslint/scope-manager/-/scope-manager-5.52.0.tgz",
-      "integrity": "sha512-AR7sxxfBKiNV0FWBSARxM8DmNxrwgnYMPwmpkC1Pl1n+eT8/I2NAUPuwDy/FmDcC6F8pBfmOcaxcxRHspgOBMw==",
+      "version": "5.54.0",
+      "resolved": "https://registry.npmjs.org/@typescript-eslint/scope-manager/-/scope-manager-5.54.0.tgz",
+      "integrity": "sha512-VTPYNZ7vaWtYna9M4oD42zENOBrb+ZYyCNdFs949GcN8Miwn37b8b7eMj+EZaq7VK9fx0Jd+JhmkhjFhvnovhg==",
       "dev": true,
       "requires": {
-        "@typescript-eslint/types": "5.52.0",
-        "@typescript-eslint/visitor-keys": "5.52.0"
+        "@typescript-eslint/types": "5.54.0",
+        "@typescript-eslint/visitor-keys": "5.54.0"
       }
     },
     "@typescript-eslint/type-utils": {
@@ -31253,19 +31253,19 @@
       }
     },
     "@typescript-eslint/types": {
-      "version": "5.52.0",
-      "resolved": "https://registry.npmjs.org/@typescript-eslint/types/-/types-5.52.0.tgz",
-      "integrity": "sha512-oV7XU4CHYfBhk78fS7tkum+/Dpgsfi91IIDy7fjCyq2k6KB63M6gMC0YIvy+iABzmXThCRI6xpCEyVObBdWSDQ==",
+      "version": "5.54.0",
+      "resolved": "https://registry.npmjs.org/@typescript-eslint/types/-/types-5.54.0.tgz",
+      "integrity": "sha512-nExy+fDCBEgqblasfeE3aQ3NuafBUxZxgxXcYfzYRZFHdVvk5q60KhCSkG0noHgHRo/xQ/BOzURLZAafFpTkmQ==",
       "dev": true
     },
     "@typescript-eslint/typescript-estree": {
-      "version": "5.52.0",
-      "resolved": "https://registry.npmjs.org/@typescript-eslint/typescript-estree/-/typescript-estree-5.52.0.tgz",
-      "integrity": "sha512-WeWnjanyEwt6+fVrSR0MYgEpUAuROxuAH516WPjUblIrClzYJj0kBbjdnbQXLpgAN8qbEuGywiQsXUVDiAoEuQ==",
+      "version": "5.54.0",
+      "resolved": "https://registry.npmjs.org/@typescript-eslint/typescript-estree/-/typescript-estree-5.54.0.tgz",
+      "integrity": "sha512-X2rJG97Wj/VRo5YxJ8Qx26Zqf0RRKsVHd4sav8NElhbZzhpBI8jU54i6hfo9eheumj4oO4dcRN1B/zIVEqR/MQ==",
       "dev": true,
       "requires": {
-        "@typescript-eslint/types": "5.52.0",
-        "@typescript-eslint/visitor-keys": "5.52.0",
+        "@typescript-eslint/types": "5.54.0",
+        "@typescript-eslint/visitor-keys": "5.54.0",
         "debug": "^4.3.4",
         "globby": "^11.1.0",
         "is-glob": "^4.0.3",
@@ -31349,12 +31349,12 @@
       }
     },
     "@typescript-eslint/visitor-keys": {
-      "version": "5.52.0",
-      "resolved": "https://registry.npmjs.org/@typescript-eslint/visitor-keys/-/visitor-keys-5.52.0.tgz",
-      "integrity": "sha512-qMwpw6SU5VHCPr99y274xhbm+PRViK/NATY6qzt+Et7+mThGuFSl/ompj2/hrBlRP/kq+BFdgagnOSgw9TB0eA==",
+      "version": "5.54.0",
+      "resolved": "https://registry.npmjs.org/@typescript-eslint/visitor-keys/-/visitor-keys-5.54.0.tgz",
+      "integrity": "sha512-xu4wT7aRCakGINTLGeyGqDn+78BwFlggwBjnHa1ar/KaGagnmwLYmlrXIrgAaQ3AE1Vd6nLfKASm7LrFHNbKGA==",
       "dev": true,
       "requires": {
-        "@typescript-eslint/types": "5.52.0",
+        "@typescript-eslint/types": "5.54.0",
         "eslint-visitor-keys": "^3.3.0"
       }
     },
diff --git a/ui/package.json b/ui/package.json
index f24b57ef80..c5dd2596ce 100644
--- a/ui/package.json
+++ b/ui/package.json
@@ -114,7 +114,7 @@
     "@types/rx": "^4.1.2",
     "@types/showdown": "1.9.4",
     "@typescript-eslint/eslint-plugin": "5.53.0",
-    "@typescript-eslint/parser": "5.52.0",
+    "@typescript-eslint/parser": "5.54.0",
     "ajv-keywords": "^5.1.0",
     "assert": "^2.0.0",
     "compression-webpack-plugin": "^10.0.0",


Test patch:
diff --git a/streampipes-extensions/streampipes-processors-transformation-jvm/src/test/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/counter/TestBooleanCounterProcessor.java b/streampipes-extensions/streampipes-processors-transformation-jvm/src/test/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/counter/TestBooleanCounterProcessor.java
new file mode 100644
index 0000000000..55fc52996b
--- /dev/null
+++ b/streampipes-extensions/streampipes-processors-transformation-jvm/src/test/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/counter/TestBooleanCounterProcessor.java
@@ -0,0 +1,177 @@
+/*
+ * 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.
+ *
+ */
+
+package org.apache.streampipes.processors.transformation.jvm.processor.booloperator.counter;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.messaging.InternalEventProcessor;
+import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.runtime.EventFactory;
+import org.apache.streampipes.model.runtime.SchemaInfo;
+import org.apache.streampipes.model.runtime.SourceInfo;
+import org.apache.streampipes.model.staticproperty.MappingPropertyUnary;
+import org.apache.streampipes.model.staticproperty.OneOfStaticProperty;
+import org.apache.streampipes.test.generator.EventStreamGenerator;
+import org.apache.streampipes.test.generator.InvocationGraphGenerator;
+import org.apache.streampipes.test.generator.grounding.EventGroundingGenerator;
+import org.apache.streampipes.wrapper.routing.SpOutputCollector;
+import org.apache.streampipes.wrapper.standalone.ProcessorParams;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+@RunWith(Parameterized.class)
+public class TestBooleanCounterProcessor {
+  private static final Logger LOG = LoggerFactory.getLogger(TestBooleanCounterProcessor.class);
+
+  @org.junit.runners.Parameterized.Parameters
+  public static Iterable<Object[]> data() {
+    return Arrays.asList(new Object[][]{
+            {"Test", "BOTH", Arrays.asList(false, true), 1},
+            {"Test", "BOTH", Arrays.asList(false, true, false), 2},
+            {"Test", "BOTH", Arrays.asList(false), 0},
+            {"Test", "TRUE -> FALSE", Arrays.asList(false, true, false, false, true), 2},
+            {"Test", "TRUE -> FALSE", Arrays.asList(true, false), 1},
+            {"Test", "TRUE -> FALSE", Arrays.asList(false), 1},
+            {"Test", "FALSE -> TRUE", Arrays.asList(false), 0},
+            {"Test", "FALSE -> TRUE", Arrays.asList(false, false, true), 1},
+            {"Test", "FALSE -> TRUE", Arrays.asList(false, true, true, false), 1},
+    });
+  }
+
+  @org.junit.runners.Parameterized.Parameter
+  public String invertFieldName;
+
+  /**
+   * Defines which boolean changes should be counted
+   * 0: BOTH
+   * 1: TRUE -> FALSE
+   * 2: FALSE -> TRUE
+   */
+  @org.junit.runners.Parameterized.Parameter(1)
+  public String flankUp;
+
+  @org.junit.runners.Parameterized.Parameter(2)
+  public List<Boolean> eventBooleans;
+
+  @org.junit.runners.Parameterized.Parameter(3)
+  public Integer expectedBooleanCount;
+
+  @Test
+  public void testBooleanCounter() {
+    BooleanCounterProcessor booleanCounter = new BooleanCounterProcessor();
+    DataProcessorDescription originalGraph = booleanCounter.declareModel();
+    originalGraph.setSupportedGrounding(EventGroundingGenerator.makeDummyGrounding());
+
+    DataProcessorInvocation graph =
+            InvocationGraphGenerator.makeEmptyInvocation(originalGraph);
+
+    graph.setInputStreams(Collections
+            .singletonList(EventStreamGenerator
+                    .makeStreamWithProperties(Collections.singletonList(invertFieldName))));
+
+    graph.setOutputStream(EventStreamGenerator.makeStreamWithProperties(Collections.singletonList(invertFieldName)));
+
+    graph.getOutputStream().getEventGrounding().getTransportProtocol().getTopicDefinition()
+            .setActualTopicName("output-topic");
+
+    graph.getStaticProperties().stream()
+            .filter(p -> p instanceof MappingPropertyUnary)
+            .map((p -> (MappingPropertyUnary) p))
+            .filter(p -> p.getInternalName().equals(BooleanCounterProcessor.FIELD_ID))
+            .findFirst().get().setSelectedProperty("s0::" + invertFieldName);
+    ProcessorParams params = new ProcessorParams(graph);
+    params.extractor().getStaticPropertyByName("flank", OneOfStaticProperty.class).getOptions()
+            .stream().filter(ot -> ot.getName().equals(flankUp)).findFirst()
+            .get().setSelected(true);
+
+
+    SpOutputCollector spOut = new SpOutputCollector() {
+      @Override
+      public void collect(Event event) {}
+      @Override
+      public void registerConsumer(String routeId, InternalEventProcessor<Map<String, Object>> consumer) {}
+      @Override
+      public void unregisterConsumer(String routeId) {}
+      @Override
+      public void connect() throws SpRuntimeException {}
+      @Override
+      public void disconnect() throws SpRuntimeException {}
+    };
+
+    booleanCounter.onInvocation(params, spOut, null);
+    Integer counter = sendEvents(booleanCounter, spOut);
+    LOG.info("Expected match count is {}", expectedBooleanCount);
+    LOG.info("Actual match count is {}", counter);
+    assertEquals(expectedBooleanCount, counter);
+  }
+
+
+  private Integer sendEvents(BooleanCounterProcessor booleanCounter, SpOutputCollector spOut) {
+    int counter = 0;
+    List<Event> events = makeEvents();
+    for (Event event : events) {
+      LOG.info("Sending event with value " + event.getFieldBySelector("s0::" + invertFieldName));
+      booleanCounter.onEvent(event, spOut);
+      try {
+        Thread.sleep(100);
+      } catch (InterruptedException e) {
+        e.printStackTrace();
+      }
+      try {
+        counter = event.getFieldBySelector(BooleanCounterProcessor.COUNT_FIELD_RUNTIME_NAME)
+                .getAsPrimitive()
+                .getAsInt();
+      } catch (IllegalArgumentException e) {
+
+      }
+    }
+
+    return counter;
+  }
+
+  private List<Event> makeEvents() {
+    List<Event> events = new ArrayList<>();
+    for (Boolean eventSetting : eventBooleans) {
+      events.add(makeEvent(eventSetting));
+    }
+    return events;
+  }
+
+  private Event makeEvent(Boolean value) {
+    Map<String, Object> map = new HashMap<>();
+    map.put(invertFieldName, value);
+    return EventFactory.fromMap(map,
+            new SourceInfo("test" + "-topic", "s0"),
+            new SchemaInfo(null, new ArrayList<>()));
+  }
+}
\ No newline at end of file
diff --git a/streampipes-extensions/streampipes-processors-transformation-jvm/src/test/java/org/apache/streampipes/processors/transformation/jvm/processors/booloperator/inverter/TestBooleanInverterProcessor.java b/streampipes-extensions/streampipes-processors-transformation-jvm/src/test/java/org/apache/streampipes/processors/transformation/jvm/processors/booloperator/inverter/TestBooleanInverterProcessor.java
new file mode 100644
index 0000000000..2a46a4308b
--- /dev/null
+++ b/streampipes-extensions/streampipes-processors-transformation-jvm/src/test/java/org/apache/streampipes/processors/transformation/jvm/processors/booloperator/inverter/TestBooleanInverterProcessor.java
@@ -0,0 +1,159 @@
+/*
+ * 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.
+ *
+ */
+
+package org.apache.streampipes.processors.transformation.jvm.processor.booloperator.inverter;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.messaging.InternalEventProcessor;
+import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.runtime.EventFactory;
+import org.apache.streampipes.model.runtime.SchemaInfo;
+import org.apache.streampipes.model.runtime.SourceInfo;
+import org.apache.streampipes.model.staticproperty.MappingPropertyUnary;
+import org.apache.streampipes.test.generator.EventStreamGenerator;
+import org.apache.streampipes.test.generator.InvocationGraphGenerator;
+import org.apache.streampipes.test.generator.grounding.EventGroundingGenerator;
+import org.apache.streampipes.wrapper.routing.SpOutputCollector;
+import org.apache.streampipes.wrapper.standalone.ProcessorParams;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+@RunWith(Parameterized.class)
+public class TestBooleanInverterProcessor {
+
+  private static final Logger LOG = LoggerFactory.getLogger(TestBooleanInverterProcessor.class);
+
+  @org.junit.runners.Parameterized.Parameters
+  public static Iterable<Object[]> data() {
+    return Arrays.asList(new Object[][]{
+        {"Test", Arrays.asList(false, true), false},
+        {"Test", Arrays.asList(false, true, false), true},
+        {"Test", Arrays.asList(false), true},
+        {"Test", Arrays.asList(false, true, false, false, true), false},
+        {"Test", Arrays.asList(true, false), true},
+        {"Test", Arrays.asList(true), false},
+    });
+  }
+
+  @org.junit.runners.Parameterized.Parameter
+  public String invertFieldName;
+
+  @org.junit.runners.Parameterized.Parameter(1)
+  public List<Boolean> eventBooleans;
+
+  @org.junit.runners.Parameterized.Parameter(2)
+  public Boolean expectedBooleanCount;
+
+  @Test
+  public void testBoolenInverter() {
+    BooleanInverterProcessor bip = new BooleanInverterProcessor();
+    DataProcessorDescription originalGraph =  bip.declareModel();
+    originalGraph.setSupportedGrounding(EventGroundingGenerator.makeDummyGrounding());
+
+    DataProcessorInvocation graph =
+        InvocationGraphGenerator.makeEmptyInvocation(originalGraph);
+
+    graph.setInputStreams(Collections
+        .singletonList(EventStreamGenerator
+            .makeStreamWithProperties(Collections.singletonList(invertFieldName))));
+
+    graph.setOutputStream(EventStreamGenerator.makeStreamWithProperties(Collections.singletonList(invertFieldName)));
+
+    graph.getOutputStream().getEventGrounding().getTransportProtocol().getTopicDefinition()
+        .setActualTopicName("output-topic");
+
+    graph.getStaticProperties().stream()
+        .filter(p -> p instanceof MappingPropertyUnary)
+        .map((p -> (MappingPropertyUnary) p))
+        .filter(p -> p.getInternalName().equals(BooleanInverterProcessor.INVERT_FIELD_ID))
+        .findFirst().get().setSelectedProperty("s0::" + invertFieldName);
+    ProcessorParams params = new ProcessorParams(graph);
+    SpOutputCollector spOut = new SpOutputCollector() {
+        @Override
+        public void collect(Event event) {}
+
+        @Override
+        public void registerConsumer(String routeId, InternalEventProcessor<Map<String, Object>> consumer) {}
+
+        @Override
+        public void unregisterConsumer(String routeId) {}
+
+        @Override
+        public void connect() throws SpRuntimeException {}
+
+        @Override
+        public void disconnect() throws SpRuntimeException {}
+    };
+
+    bip.onInvocation(params, spOut, null);
+
+    boolean result = sendEvents(bip, spOut);
+
+    LOG.info("Expected boolean is {}", expectedBooleanCount);
+    LOG.info("Actual boolean is {}", result);
+    assertEquals(expectedBooleanCount, result);
+  }
+
+
+  private boolean sendEvents(BooleanInverterProcessor trend, SpOutputCollector spOut) {
+    boolean result = false;
+    List<Event> events = makeEvents();
+    for (Event event : events) {
+      LOG.info("Sending event with value " + event.getFieldBySelector("s0::" + invertFieldName));
+      trend.onEvent(event, spOut);
+      try {
+        Thread.sleep(100);
+      } catch (InterruptedException e) {
+        e.printStackTrace();
+      }
+      result = event.getFieldBySelector("s0::" + invertFieldName).getAsPrimitive().getAsBoolean();
+    }
+
+    return result;
+  }
+
+  private List<Event> makeEvents() {
+    List<Event> events = new ArrayList<>();
+    for (Boolean eventSetting : eventBooleans) {
+      events.add(makeEvent(eventSetting));
+    }
+    return events;
+  }
+
+  private Event makeEvent(Boolean value) {
+    Map<String, Object> map = new HashMap<>();
+    map.put(invertFieldName, value);
+    return EventFactory.fromMap(map, new SourceInfo("test" + "-topic", "s0"),
+            new SchemaInfo(null, new ArrayList<>()));
+  }
+}
\ No newline at end of file