Repository: apache/streampipes (java)

Issue:
Cannot deploy a pipeline with Trend data processor
### Apache StreamPipes version

dev (current development state)

### Affected StreamPipes components

Backend

### What happened?

Error while trying to deploy a Trend Data Processor:
Error between @ Line: 1. Position: 0 and @ Line: 1. Position: 80. Syntax error in SiddhiQL, extraneous input 'Quality' expecting {STRING, INT, LONG, FLOAT, DOUBLE, BOOL, OBJECT}.

### How to reproduce?

_No response_

### Expected behavior

_No response_

### Additional technical information

_No response_

### Are you willing to submit a PR?

Yes, I am willing to submit a PR!

Pull request:
fix(#2836): Sanitize blank space in Siddhi queries
<!--
  ~ 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.
-->

Fix #2836

### Remarks
<!--
Is there anything left we need to pay attention on?
Are there some references that might be important? E.g. links to Confluence, or discussions
on the mailing list or GitHub.
-->
PR introduces (a) breaking change(s): no

PR introduces (a) deprecation(s): no


Patch:
diff --git a/streampipes-client-go/streampipes/functions_api.go b/streampipes-client-go/streampipes/functions_api.go
new file mode 100644
index 0000000000..9d7e904555
--- /dev/null
+++ b/streampipes-client-go/streampipes/functions_api.go
@@ -0,0 +1,151 @@
+//
+// 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 streampipes
+
+import (
+	"github.com/apache/streampipes/streampipes-client-go/streampipes/config"
+	"github.com/apache/streampipes/streampipes-client-go/streampipes/internal/serializer"
+	"github.com/apache/streampipes/streampipes-client-go/streampipes/internal/util"
+	"github.com/apache/streampipes/streampipes-client-go/streampipes/model/functions"
+	"io"
+	"log"
+	"net/http"
+)
+
+type Functions struct {
+	endpoint
+}
+
+func NewFunctions(clientConfig config.StreamPipesClientConfig) *Functions {
+
+	return &Functions{
+		endpoint{config: clientConfig},
+	}
+}
+
+func (f *Functions) GetFunctionLogs(functionId string) ([]functions.SpLogEntry, error) {
+
+	endPointUrl := util.NewStreamPipesApiPath(f.config.Url, "streampipes-backend/api/v2/functions", []string{functionId, "logs"})
+	log.Printf("Get data from: %s", endPointUrl)
+
+	response, err := f.executeRequest("GET", endPointUrl)
+	if err != nil {
+		return nil, err
+	}
+
+	if response.StatusCode != http.StatusOK {
+		err = f.handleStatusCode(response)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	body, err := io.ReadAll(response.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	unmarshalData, err := serializer.NewSpLogEntriesDeserializer().Unmarshal(body)
+	if err != nil {
+		return nil, err
+	}
+	spLogEntry := unmarshalData.([]functions.SpLogEntry)
+
+	return spLogEntry, nil
+}
+
+func (f *Functions) GetFunctionMetrics(functionId string) (functions.SpMetricsEntry, error) {
+
+	endPointUrl := util.NewStreamPipesApiPath(f.config.Url, "streampipes-backend/api/v2/functions", []string{functionId, "metrics"})
+	log.Printf("Get data from: %s", endPointUrl)
+
+	response, err := f.executeRequest("GET", endPointUrl)
+	if err != nil {
+		return functions.SpMetricsEntry{}, err
+	}
+
+	if response.StatusCode != http.StatusOK {
+		err = f.handleStatusCode(response)
+		if err != nil {
+			return functions.SpMetricsEntry{}, err
+		}
+	}
+
+	body, err := io.ReadAll(response.Body)
+	if err != nil {
+		return functions.SpMetricsEntry{}, err
+	}
+
+	unmarshalData, err := serializer.NewSpMetricsEntryDeserializer().Unmarshal(body)
+	if err != nil {
+		return functions.SpMetricsEntry{}, err
+	}
+	spLogEntry := unmarshalData.(functions.SpMetricsEntry)
+
+	return spLogEntry, nil
+}
+
+func (f *Functions) GetAllFunction() ([]functions.FunctionDefinition, error) {
+
+	endPointUrl := util.NewStreamPipesApiPath(f.config.Url, "streampipes-backend/api/v2/functions", nil)
+	log.Printf("Get data from: %s", endPointUrl)
+
+	response, err := f.executeRequest("GET", endPointUrl)
+	if err != nil {
+		return nil, err
+	}
+
+	if response.StatusCode != http.StatusOK {
+		err = f.handleStatusCode(response)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	body, err := io.ReadAll(response.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	unmarshalData, err := serializer.NewFunctionDefinitionsDeserializer().Unmarshal(body)
+	if err != nil {
+		return nil, err
+	}
+	functionDefinition := unmarshalData.([]functions.FunctionDefinition)
+
+	return functionDefinition, nil
+}
+
+func (f *Functions) DeleteSingleFunction(functionId string) error {
+	endPointUrl := util.NewStreamPipesApiPath(f.config.Url, "streampipes-backend/api/v2/functions", []string{functionId})
+	log.Printf("Delete data from: %s", endPointUrl)
+
+	response, err := f.executeRequest("DELETE", endPointUrl)
+	if err != nil {
+		return err
+	}
+
+	if response.StatusCode != http.StatusOK {
+		err = f.handleStatusCode(response)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
diff --git a/streampipes-client-go/streampipes/internal/serializer/deserializer.go b/streampipes-client-go/streampipes/internal/serializer/deserializer.go
index 1ef7f7d1dd..fcdd7593cf 100644
--- a/streampipes-client-go/streampipes/internal/serializer/deserializer.go
+++ b/streampipes-client-go/streampipes/internal/serializer/deserializer.go
@@ -20,7 +20,11 @@ package serializer
 import (
 	"encoding/json"
 	"github.com/apache/streampipes/streampipes-client-go/streampipes/model/data_lake"
+
+	"github.com/apache/streampipes/streampipes-client-go/streampipes/model/functions"
+
 	"github.com/apache/streampipes/streampipes-client-go/streampipes/model/streampipes_user"
+
 	"github.com/apache/streampipes/streampipes-client-go/streampipes/model/streampipes_version"
 )
 
@@ -92,6 +96,53 @@ func (d *StreamPipesVersionDeserializer) Unmarshal(data []byte) (interface{}, er
 	return dataSeries, nil
 }
 
+
+type SpLogEntriesDeserializer struct{}
+
+func NewSpLogEntriesDeserializer() *SpLogEntriesDeserializer {
+	return &SpLogEntriesDeserializer{}
+}
+
+func (p *SpLogEntriesDeserializer) Unmarshal(data []byte) (interface{}, error) {
+	var userAccount []functions.SpLogEntry
+	err := json.Unmarshal(data, &userAccount)
+	if err != nil {
+		return nil, err
+	}
+	return userAccount, nil
+}
+
+type SpMetricsEntryDeserializer struct{}
+
+func NewSpMetricsEntryDeserializer() *SpMetricsEntryDeserializer {
+	return &SpMetricsEntryDeserializer{}
+}
+
+func (p *SpMetricsEntryDeserializer) Unmarshal(data []byte) (interface{}, error) {
+	var spMetricsEntry functions.SpMetricsEntry
+	err := json.Unmarshal(data, &spMetricsEntry)
+	if err != nil {
+		return nil, err
+	}
+	return spMetricsEntry, nil
+}
+
+type FunctionDefinitionsDeserializer struct{}
+
+func NewFunctionDefinitionsDeserializer() *FunctionDefinitionsDeserializer {
+	return &FunctionDefinitionsDeserializer{}
+}
+
+func (p *FunctionDefinitionsDeserializer) Unmarshal(data []byte) (interface{}, error) {
+	var functionDefinitions []functions.FunctionDefinition
+	err := json.Unmarshal(data, &functionDefinitions)
+	if err != nil {
+		return nil, err
+	}
+	return functionDefinitions, nil
+  
+}
+
 type ShortUserInfosDeserializer struct{}
 
 func NewShortUserInfosDeserializer() *ShortUserInfosDeserializer {
@@ -120,4 +171,5 @@ func (p *UserAccountDeserializer) Unmarshal(data []byte) (interface{}, error) {
 		return nil, err
 	}
 	return userAccount, nil
+
 }
diff --git a/streampipes-client-go/streampipes/model/functions/functions.go b/streampipes-client-go/streampipes/model/functions/functions.go
new file mode 100644
index 0000000000..c360c221d9
--- /dev/null
+++ b/streampipes-client-go/streampipes/model/functions/functions.go
@@ -0,0 +1,52 @@
+//
+// 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 functions
+
+type FunctionId struct {
+	Id      string `json:"id"`
+	Version int    `json:"version"`
+}
+
+type FunctionDefinition struct {
+	FunctionIds     FunctionId `json:"functionId"`
+	ConsumedStreams []string   `json:"consumedStreams"`
+}
+
+type SpLogEntry struct {
+	Timestamp    int64        `json:"timestamp"`
+	ErrorMessage SpLogMessage `json:"errorMessage"`
+}
+
+type SpLogMessage struct {
+	Level          string `json:"level"`
+	Title          string `json:"title"`
+	Detail         string `json:"detail"`
+	Cause          string `json:"cause"`
+	FullStackTrace string `json:"fullStackTrace"`
+}
+
+type SpMetricsEntry struct {
+	LastTimestamp int64                     `json:"lastTimestamp"`
+	MessagesIn    map[string]MessageCounter `json:"messagesIn"`
+	MessagesOut   MessageCounter            `json:"messagesOut"`
+}
+
+type MessageCounter struct {
+	LastTimestamp int64 `json:"lastTimestamp"`
+	Counter       int64 `json:"counter"`
+}
diff --git a/streampipes-client-go/streampipes/streampipes_client.go b/streampipes-client-go/streampipes/streampipes_client.go
index c968a4ee4f..7741fcc9f3 100644
--- a/streampipes-client-go/streampipes/streampipes_client.go
+++ b/streampipes-client-go/streampipes/streampipes_client.go
@@ -72,7 +72,13 @@ func (s *StreamPipesClient) StreamPipesVersion() *Versions {
 	return NewVersions(s.config)
 }
 
+func (s *StreamPipesClient) Function() *Functions {
+
+	return NewFunctions(s.config)
+}
+
 func (s *StreamPipesClient) UserInfo() *StreamPipesUserInfo {
 
 	return NewStreamPipesUserInfo(s.config)
+
 }
diff --git a/streampipes-client-python/poetry.lock b/streampipes-client-python/poetry.lock
index 9ef79b9cf2..f9c06b8139 100644
--- a/streampipes-client-python/poetry.lock
+++ b/streampipes-client-python/poetry.lock
@@ -505,45 +505,45 @@ files = [
 
 [[package]]
 name = "confluent-kafka"
-version = "2.3.0"
+version = "2.4.0"
 description = "Confluent's Python client for Apache Kafka"
 optional = false
 python-versions = "*"
 files = [
-    {file = "confluent-kafka-2.3.0.tar.gz", hash = "sha256:4069e7b56e0baf9db18c053a605213f0ab2d8f23715dca7b3bd97108df446ced"},
-    {file = "confluent_kafka-2.3.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:5df845755cd3ebb9165ca00fd1d3a7d514c61e84d9fcbe7babb91193fe9b369c"},
-    {file = "confluent_kafka-2.3.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:9ab2217875b731bd390582952e0f9cbe3e7b34774490f01afca70728f0d8b469"},
-    {file = "confluent_kafka-2.3.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:62046e8a75c7a6883a0f1f4a635573fd7e1665eeacace65e7f6d59cbaa94697d"},
-    {file = "confluent_kafka-2.3.0-cp310-cp310-manylinux_2_28_aarch64.whl", hash = "sha256:1eba38061e9ed1c0a369c129bf01d07499286cc3cb295398b88a7037c14371fb"},
-    {file = "confluent_kafka-2.3.0-cp310-cp310-win_amd64.whl", hash = "sha256:a6abece28598fa2b59d2b9399fcec03440aaa73fd207fdad048a6030d7e897e1"},
-    {file = "confluent_kafka-2.3.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:d55fbdcd75586dd17fe3fe64f4b4efa1c93ce9dd09c275de46f75772826e8860"},
-    {file = "confluent_kafka-2.3.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:ec17b26d6155feeaded4a435ba949095aea9699afb65309d8f22e55722f53c48"},
-    {file = "confluent_kafka-2.3.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:e9b42bf1b75fdd9aa20c77b27f166f6289440ac649f70622a0117a8e7aa6169d"},
-    {file = "confluent_kafka-2.3.0-cp311-cp311-manylinux_2_28_aarch64.whl", hash = "sha256:7f9f4099aaf2c5daae828d2f356e4277d0ef0485ec883dbe395f0c0e054450d0"},
-    {file = "confluent_kafka-2.3.0-cp311-cp311-win_amd64.whl", hash = "sha256:1c6b29d57df99dabd45e67fd0aa46f17f195b057734ad84cf9cfdc2542855c10"},
-    {file = "confluent_kafka-2.3.0-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:6b46ce75bda0c092da103dbd55cb0ba429c73c232e70b476b19a0ab247ec9057"},
-    {file = "confluent_kafka-2.3.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:af60af786a7b8cbeafea51a9416664b96b0f5ef6243172b0bc59e5f75e8bd86a"},
-    {file = "confluent_kafka-2.3.0-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:e08b601e09a584c6a4a8c323a71e92fca31a8826ed33b5b95b26783b7a996026"},
-    {file = "confluent_kafka-2.3.0-cp312-cp312-manylinux_2_28_aarch64.whl", hash = "sha256:7fd1ab257d4fa0e2a98529e4eb2102cf8352ad6b3d22110d6cf0bb1f598893d9"},
-    {file = "confluent_kafka-2.3.0-cp312-cp312-win_amd64.whl", hash = "sha256:1ccf6483d86535627cad7b94982ea95d9fa9ae04ddb552e097c1211ffcde5ea7"},
-    {file = "confluent_kafka-2.3.0-cp36-cp36m-macosx_10_9_x86_64.whl", hash = "sha256:030fb237927ec2296882a9bb96237ebf86e48388166b15ec0bbf3fdeb48df81a"},
-    {file = "confluent_kafka-2.3.0-cp36-cp36m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:bc24c57a52c206648685e1c536afb8447d1cbbbf3871cacebccf2e5b67bdf535"},
-    {file = "confluent_kafka-2.3.0-cp36-cp36m-manylinux_2_28_aarch64.whl", hash = "sha256:25292a9a8ef7765c85636851d6c4d5e5e98d6ead627b59637b24a5779e8a4b02"},
-    {file = "confluent_kafka-2.3.0-cp36-cp36m-win_amd64.whl", hash = "sha256:d634d4d9914b0a28ec3e37ab7b150173aa34c81fd5bd0b4dcac972b520ad56cc"},
-    {file = "confluent_kafka-2.3.0-cp37-cp37m-macosx_10_9_x86_64.whl", hash = "sha256:ebf460d90478bcd1b4564023a5b081c6e5390b28dbabbb17ee664e223830465d"},
-    {file = "confluent_kafka-2.3.0-cp37-cp37m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:cec97f8c6564b16504d30fe42c22fd4a86c406dbcd45c337b93c21e876e20628"},
-    {file = "confluent_kafka-2.3.0-cp37-cp37m-manylinux_2_28_aarch64.whl", hash = "sha256:128ddb28c19ab57c18c0e3d8209d089b6b90ff111b20108764f6798468432693"},
-    {file = "confluent_kafka-2.3.0-cp37-cp37m-win_amd64.whl", hash = "sha256:0470dc5e56e639693149961409bc6b663df94d68ceae296ae9c42e079fe65d00"},
-    {file = "confluent_kafka-2.3.0-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:b539064fef35386936a0d2dadf8a82b8b0ae325af95d9263a2431b82671c4702"},
-    {file = "confluent_kafka-2.3.0-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:4f9998f781a1da0c9dcb5506792a39799cb54e28c6f986ddc73e362887042f7c"},
-    {file = "confluent_kafka-2.3.0-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:f175e11facaf12130abd5d2d471db39d7cc89126c4d991527cf14e3da22c635c"},
-    {file = "confluent_kafka-2.3.0-cp38-cp38-manylinux_2_28_aarch64.whl", hash = "sha256:f9842720ed0debcf4620710e01d356681a4812441f1ff49664fc205d1f9120e5"},
-    {file = "confluent_kafka-2.3.0-cp38-cp38-win_amd64.whl", hash = "sha256:cf015e547b82a74a87d7363d0d42e4cd0ca23b01cdb479639a340f385581ea04"},
-    {file = "confluent_kafka-2.3.0-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:e5c740ead14a2510e15f63e67b19d48ae48a7f30ef4823d5af125bad528033d1"},
-    {file = "confluent_kafka-2.3.0-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:6ae5e6a6dcd5ce85b9153c21c9f0b83e0cc88a5955b5334079db76c2267deb63"},
-    {file = "confluent_kafka-2.3.0-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:ca36a8d1d49fd55cca1b7ec3090ca2684a933e63f196f0e3e506194b189fc31e"},
-    {file = "confluent_kafka-2.3.0-cp39-cp39-manylinux_2_28_aarch64.whl", hash = "sha256:210f2d346d1006e9b95c5204f7255735d4cb5ec962a3d1a68ac60c02e2763ae4"},
-    {file = "confluent_kafka-2.3.0-cp39-cp39-win_amd64.whl", hash = "sha256:cb279e369121e07ccb419220fc039127345a9e5f72f4abf7dda0e2e06a12b604"},
+    {file = "confluent-kafka-2.4.0.tar.gz", hash = "sha256:201c6182304c5864c8a1b1aae2f99bf51fa332017abbec05a3a1fb2ff242c41d"},
+    {file = "confluent_kafka-2.4.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:827820321e0ac6a7f75f119f7ab4dcbbbd59c95d7807f6eb79f03a4cb6b9ab25"},
+    {file = "confluent_kafka-2.4.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:4c74a06a658f72dd0f6df9b0fbfc5b1b483b3263fde6abd925bba512879995cc"},
+    {file = "confluent_kafka-2.4.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:108bc30e1153734cd36d88a299c7cf0c69fe6c95d67f4cce9f9180adaefef839"},
+    {file = "confluent_kafka-2.4.0-cp310-cp310-manylinux_2_28_aarch64.whl", hash = "sha256:cbe32df72b9d7fee198cf12eb3770270d99b122613b0e3abfb55a13294f6a54a"},
+    {file = "confluent_kafka-2.4.0-cp310-cp310-win_amd64.whl", hash = "sha256:c2e9e1dd2238e5b55cf541c9dbd3e665531418e0cc965e6143f593f64a3f79a5"},
+    {file = "confluent_kafka-2.4.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:a28bf5b59f28e3357d113cc8c50ba09389dcc3327ca13cc502e9c3228f2ab5dc"},
+    {file = "confluent_kafka-2.4.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:506c867d987ab78494fc94ec6a966ddcb3ce3cb75809e098060d654d280281e0"},
+    {file = "confluent_kafka-2.4.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:3b02cd0eb338a8dc5db0ea07243bbf457670655a3846732f81a091b8e3d5d456"},
+    {file = "confluent_kafka-2.4.0-cp311-cp311-manylinux_2_28_aarch64.whl", hash = "sha256:15d356dd1c7699ed332f262bf0aa825a177c50f7debca0f1ec14873531603b2d"},
+    {file = "confluent_kafka-2.4.0-cp311-cp311-win_amd64.whl", hash = "sha256:0c1b89e1bd790dd82b64a4dd746e7a0ce3ebed7f17b0642d488c1bb0056ba763"},
+    {file = "confluent_kafka-2.4.0-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:a11acfd79f0448095c6492d2ea34cd3df4864d78d72326bed0859d875eb7a62d"},
+    {file = "confluent_kafka-2.4.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:dd53d58a6648e4d5e435ff5178248b91af050a4478ed51a95b2b0eaedddafdb1"},
+    {file = "confluent_kafka-2.4.0-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:8a5e9cc9fff71c70f1520ab917f33cf66181b29c46cf9940da1af6a52cc086a6"},
+    {file = "confluent_kafka-2.4.0-cp312-cp312-manylinux_2_28_aarch64.whl", hash = "sha256:b0d5a2d7563b9ca9bef09668d6aae6e360f1e9455672dc0886d78d76ced682c9"},
+    {file = "confluent_kafka-2.4.0-cp312-cp312-win_amd64.whl", hash = "sha256:d09aa0c67532c47ec6cd08c50feaa8338e77d18fc83181c46e5a5cb88a5b25ce"},
+    {file = "confluent_kafka-2.4.0-cp36-cp36m-macosx_10_9_x86_64.whl", hash = "sha256:f2e39794270088157f65e5b2b5ba7e0217759412de20088f5873bdb176084ab3"},
+    {file = "confluent_kafka-2.4.0-cp36-cp36m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:0591645e709ea8392f8d41b9ff752f1cdb57c71730c9be2deae3184303d3efaf"},
+    {file = "confluent_kafka-2.4.0-cp36-cp36m-manylinux_2_28_aarch64.whl", hash = "sha256:07f38339da116527d9516a0b1899b499ebfe5f906ef54c0780a309ebee585c12"},
+    {file = "confluent_kafka-2.4.0-cp36-cp36m-win_amd64.whl", hash = "sha256:c661f4c014f78ee478667a5018fb62ec2a4b61322410a9be314bdbf6ead481d7"},
+    {file = "confluent_kafka-2.4.0-cp37-cp37m-macosx_10_9_x86_64.whl", hash = "sha256:a8b440684a22d9239b857d1be065ea2844537ff20bb675c75e7a9e2adbd5ce9a"},
+    {file = "confluent_kafka-2.4.0-cp37-cp37m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:9ac7120919a4189018120a40da8d9c05bd6e024f60638901483bb6a46c45a115"},
+    {file = "confluent_kafka-2.4.0-cp37-cp37m-manylinux_2_28_aarch64.whl", hash = "sha256:6a2b81902b0c824549f85f49774535285242b1472e52ab0a18260526840a8ecf"},
+    {file = "confluent_kafka-2.4.0-cp37-cp37m-win_amd64.whl", hash = "sha256:78f6776aca06c81308c8be66c754a3b8281a20dd9af697b2cfcb5c2272878df9"},
+    {file = "confluent_kafka-2.4.0-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:2858fb428f29086df2a50554d9093d176ec4b1f76240a62efe136acc493f7fb6"},
+    {file = "confluent_kafka-2.4.0-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:9608b8b1145a0b41e6c6949ac9ec974e74f1ed2d065b26eae73e208668fa20cf"},
+    {file = "confluent_kafka-2.4.0-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:baaf1022972bd213bf20cfccb1b13b613d7ab13ebcafacd65220bd5908f5e460"},
+    {file = "confluent_kafka-2.4.0-cp38-cp38-manylinux_2_28_aarch64.whl", hash = "sha256:16be3360f040b5d17dcb7e1b66a170b72f428c98d0978705b0a3081de756b2c7"},
+    {file = "confluent_kafka-2.4.0-cp38-cp38-win_amd64.whl", hash = "sha256:45e6f719f6bd95210e0451b50cd19f33a44d7b16be560217240caa71f165420d"},
+    {file = "confluent_kafka-2.4.0-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:3cf1b40ae59569e50ba15da1dc0030bac904b74d8aff82f02a16df3bb9b87e53"},
+    {file = "confluent_kafka-2.4.0-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:5b5dd225596325159eee928caba47e0cf3f03ce3c8244371457a1823d7c66c22"},
+    {file = "confluent_kafka-2.4.0-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:d338e8189f5836b916c632e0f2db9a197352c199db7323ca3f2bd1785cb9a747"},
+    {file = "confluent_kafka-2.4.0-cp39-cp39-manylinux_2_28_aarch64.whl", hash = "sha256:40d18f1bd1bac6d2cf8dbf47da11cdb82d077b26a81ea551136515a1079ec532"},
+    {file = "confluent_kafka-2.4.0-cp39-cp39-win_amd64.whl", hash = "sha256:52f937ab1cde9aacb01288293d374bf09da728b79cf907504dec8f3c4f82030e"},
 ]
 
 [package.extras]
diff --git a/streampipes-extensions/streampipes-processors-filters-siddhi/pom.xml b/streampipes-extensions/streampipes-processors-filters-siddhi/pom.xml
index 884581a1cd..5e7493d707 100644
--- a/streampipes-extensions/streampipes-processors-filters-siddhi/pom.xml
+++ b/streampipes-extensions/streampipes-processors-filters-siddhi/pom.xml
@@ -52,6 +52,11 @@
             <artifactId>junit-jupiter-api</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-params</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
             <artifactId>streampipes-test-utils</artifactId>
diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/generator/SiddhiAppGenerator.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/generator/SiddhiAppGenerator.java
index 4838f16a56..9b70a74a49 100644
--- a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/generator/SiddhiAppGenerator.java
+++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/engine/generator/SiddhiAppGenerator.java
@@ -78,7 +78,7 @@ public class SiddhiAppGenerator {
         .getQueries()
         .forEach(query -> this.siddhiAppString.append(query).append("\n"));
 
-    LOG.info("Registering statement: \n" + this.siddhiAppString);
+    LOG.info("Registering statement: \n{}", this.siddhiAppString);
 
   }
 }
diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/model/EventPropertyDef.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/model/EventPropertyDef.java
index 175f53e2c1..4d59c1fe8e 100644
--- a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/model/EventPropertyDef.java
+++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/model/EventPropertyDef.java
@@ -19,10 +19,16 @@ package org.apache.streampipes.wrapper.siddhi.model;
 
 public class EventPropertyDef {
 
+  public static final String WHITESPACE_REPLACEMENT = "__w__";
+
   private String selectorPrefix;
   private final String fieldName;
   private final String fieldType;
 
+  public static String toOriginalFieldName(String sanitizedFieldName) {
+    return sanitizedFieldName.replaceAll(WHITESPACE_REPLACEMENT, " ");
+  }
+
   public EventPropertyDef(String fieldName,
                           String fieldType) {
     this.fieldName = fieldName;
@@ -40,7 +46,7 @@ public class EventPropertyDef {
   }
 
   public String getFieldName() {
-    return fieldName;
+    return fieldName.replaceAll(" ", WHITESPACE_REPLACEMENT);
   }
 
   public String getFieldType() {
diff --git a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/utils/SiddhiUtils.java b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/utils/SiddhiUtils.java
index 041bbce894..e2e589dc41 100644
--- a/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/utils/SiddhiUtils.java
+++ b/streampipes-wrapper-siddhi/src/main/java/org/apache/streampipes/wrapper/siddhi/utils/SiddhiUtils.java
@@ -19,10 +19,12 @@ package org.apache.streampipes.wrapper.siddhi.utils;
 
 
 import org.apache.streampipes.extensions.api.pe.param.IDataProcessorParameters;
+import org.apache.streampipes.model.constants.PropertySelectorConstants;
 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.wrapper.siddhi.constants.SiddhiConstants;
+import org.apache.streampipes.wrapper.siddhi.model.EventPropertyDef;
 
 import io.siddhi.core.event.Event;
 import io.siddhi.query.api.definition.Attribute;
@@ -68,7 +70,7 @@ public class SiddhiUtils {
         outputKey = outputKey.substring(2);
       }
       Object data = event.getData(i);
-      outMap.put(outputKey, data);
+      outMap.put(EventPropertyDef.toOriginalFieldName(outputKey), data);
     }
 
     return outMap;
@@ -101,11 +103,13 @@ public class SiddhiUtils {
     return eventName
         .replaceAll("\\.", "")
         .replaceAll("-", "")
-        .replaceAll("::", "");
+        .replaceAll(PropertySelectorConstants.PROPERTY_DELIMITER, "");
   }
 
   public static String prepareProperty(String propertyName) {
-    return propertyName.replaceAll("::", "");
+    return propertyName
+        .replaceAll(PropertySelectorConstants.PROPERTY_DELIMITER, "")
+        .replaceAll(" ", EventPropertyDef.WHITESPACE_REPLACEMENT);
   }
 
 }


Test patch:
diff --git a/streampipes-extensions/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java b/streampipes-extensions/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java
index a77552bfbb..9b96316a35 100644
--- a/streampipes-extensions/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java
+++ b/streampipes-extensions/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java
@@ -17,141 +17,161 @@
  */
 package org.apache.streampipes.processors.siddhi.trend;
 
-//@RunWith(Parameterized.class)
+
+import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.output.CustomOutputStrategy;
+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.sdk.helpers.Tuple2;
+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.params.generator.DataProcessorParameterGenerator;
+import org.apache.streampipes.wrapper.siddhi.engine.callback.SiddhiDebugCallback;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+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 java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+
 public class TestTrendProcessor {
-//
-//  private static final Logger LOG = LoggerFactory.getLogger(TrendProcessor.class);
-//
-//  @org.junit.runners.Parameterized.Parameters
-//  public static Iterable<Object[]> data() {
-//    return Arrays.asList(new Object[][]{
-//        {1, 100, TrendOperator.INCREASE, Arrays.asList(new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 2)), 1},
-//        {1, 100, TrendOperator.INCREASE, Arrays.asList(
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 2))
-//            , 1},
-//        {1, 100, TrendOperator.DECREASE, Arrays.asList(
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 0))
-//            , 1},
-//        {1, 100, TrendOperator.INCREASE, Arrays.asList(
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 2),
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 2))
-//            , 2},
-//        {1, 200, TrendOperator.INCREASE, Arrays.asList(
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 2),
-//            new Tuple2<>(100, 1),
-//            new Tuple2<>(100, 2))
-//            , 0},
-//
-//    });
-//  }
-//
-//  @org.junit.runners.Parameterized.Parameter
-//  public Integer timeWindow;
-//
-//  @org.junit.runners.Parameterized.Parameter(1)
-//  public Integer increase;
-//
-//  @org.junit.runners.Parameterized.Parameter(2)
-//  public TrendOperator trendOperator;
-//
-//  @org.junit.runners.Parameterized.Parameter(3)
-//  public List<Tuple2<Integer, Integer>> eventSettings;
-//
-//  @org.junit.runners.Parameterized.Parameter(4)
-//  public Integer expectedMatchCount;
-//
-//  @Test
-//  public void testTrend() {
-//    final Integer[] actualMatchCount = {0};
-//    DataProcessorDescription originalGraph = new TrendProcessor().declareModel();
-//    originalGraph.setSupportedGrounding(EventGroundingGenerator.makeDummyGrounding());
-//
-//    DataProcessorInvocation graph =
-//        InvocationGraphGenerator.makeEmptyInvocation(originalGraph);
-//
-//    graph.setInputStreams(Collections
-//        .singletonList(EventStreamGenerator
-//            .makeStreamWithProperties(Collections.singletonList("randomValue"))));
-//
-//    graph.setOutputStrategies(
-//        graph.getOutputStrategies()
-//            .stream()
-//            .filter(o -> o instanceof CustomOutputStrategy)
-//            .peek(o -> ((CustomOutputStrategy) o).setSelectedPropertyKeys(Arrays.asList("s0::randomValue")))
-//            .collect(Collectors.toList())
-//    );
-//
-//    graph.setOutputStream(EventStreamGenerator.makeStreamWithProperties(Collections.singletonList("randomValue")));
-//
-//    graph.getOutputStream().getEventGrounding().getTransportProtocol().getTopicDefinition()
-//        .setActualTopicName("output-topic");
-//
-//    var visitor = new TrendConfigurationVisitor("s0::randomValue",
-//        trendOperator,
-//        increase,
-//        timeWindow);
-//    graph.getStaticProperties().forEach(sp -> sp.accept(visitor));
-//
-//    var processorParams = new DataProcessorParameterGenerator().makeParameters(graph);
-//
-//    SiddhiDebugCallback callback = new SiddhiDebugCallback() {
-//      @Override
-//      public void onEvent(io.siddhi.core.event.Event event) {
-//        actualMatchCount[0]++;
-//      }
-//
-//      @Override
-//      public void onEvent(List<io.siddhi.core.event.Event> events) {
-//
-//      }
-//    };
-//
-//    TrendProcessor trend = new TrendProcessor(callback);
-//    trend.onPipelineStarted(processorParams, null, null);
-//
-//    sendEvents(trend);
-//    LOG.info("Expected match count is {}", expectedMatchCount);
-//    LOG.info("Actual match count is {}", actualMatchCount[0]);
-//    assertEquals(expectedMatchCount, actualMatchCount[0]);
-//  }
-//
-//  private void sendEvents(TrendProcessor trend) {
-//    List<Tuple2<Integer, Event>> events = makeEvents();
-//    for (Tuple2<Integer, Event> event : events) {
-//      LOG.info("Sending event with value " + event.v.getFieldBySelector("s0::randomValue"));
-//      trend.onEvent(event.v, null);
-//      try {
-//        Thread.sleep(event.k);
-//      } catch (InterruptedException e) {
-//        e.printStackTrace();
-//      }
-//    }
-//  }
-//
-//  private List<Tuple2<Integer, Event>> makeEvents() {
-//    List<Tuple2<Integer, Event>> events = new ArrayList<>();
-//    for (Tuple2<Integer, Integer> eventSetting : eventSettings) {
-//      events.add(makeEvent(eventSetting.k, eventSetting.v));
-//    }
-//    return events;
-//  }
-//
-//  private Tuple2<Integer, Event> makeEvent(Integer timeout, Integer value) {
-//    Map<String, Object> map = new HashMap<>();
-//    map.put("randomValue", value);
-//    return new Tuple2<>(timeout, EventFactory.fromMap(map, new SourceInfo("test"
-//            + "-topic", "s0"),
-//        new SchemaInfo(null,
-//            new ArrayList<>())));
-//  }
+
+  private static final Logger LOG = LoggerFactory.getLogger(TrendProcessor.class);
+
+  static Stream<Arguments> data() {
+    return Stream.of(
+        Arguments.of(1, 100, TrendOperator.INCREASE, Arrays.asList(new Tuple2<>(100, 1),
+            new Tuple2<>(100, 2)), 1),
+        Arguments.of(1, 100, TrendOperator.INCREASE, Arrays.asList(
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 2))
+            , 1),
+        Arguments.of(1, 100, TrendOperator.DECREASE, Arrays.asList(
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 0))
+            , 1),
+        Arguments.of(1, 100, TrendOperator.INCREASE, Arrays.asList(
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 2),
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 2))
+            , 2),
+        Arguments.of(1, 200, TrendOperator.INCREASE, Arrays.asList(
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 2),
+            new Tuple2<>(100, 1),
+            new Tuple2<>(100, 2))
+            , 0)
+    );
+  }
+
+  @ParameterizedTest
+  @MethodSource("data")
+  public void testTrend(int timeWindow,
+                        int increase,
+                        TrendOperator trendOperator,
+                        List<Tuple2<Integer, Integer>> eventSettings,
+                        int expectedMatchCount) {
+    final Integer[] actualMatchCount = {0};
+    DataProcessorDescription originalGraph = new TrendProcessor().declareModel();
+    originalGraph.setSupportedGrounding(EventGroundingGenerator.makeDummyGrounding());
+
+    DataProcessorInvocation graph =
+        InvocationGraphGenerator.makeEmptyInvocation(originalGraph);
+
+    graph.setInputStreams(Collections
+        .singletonList(EventStreamGenerator
+            .makeStreamWithProperties(Collections.singletonList("randomValue"))));
+
+    graph.setOutputStrategies(
+        graph.getOutputStrategies()
+            .stream()
+            .filter(o -> o instanceof CustomOutputStrategy)
+            .peek(o -> ((CustomOutputStrategy) o).setSelectedPropertyKeys(List.of("s0::randomValue")))
+            .collect(Collectors.toList())
+    );
+
+    graph.setOutputStream(EventStreamGenerator.makeStreamWithProperties(Collections.singletonList("randomValue")));
+
+    graph.getOutputStream().getEventGrounding().getTransportProtocol().getTopicDefinition()
+        .setActualTopicName("output-topic");
+
+    var visitor = new TrendConfigurationVisitor("s0::randomValue",
+        trendOperator,
+        increase,
+        timeWindow);
+    graph.getStaticProperties().forEach(sp -> sp.accept(visitor));
+
+    var processorParams = new DataProcessorParameterGenerator().makeParameters(graph);
+
+    SiddhiDebugCallback callback = new SiddhiDebugCallback() {
+      @Override
+      public void onEvent(io.siddhi.core.event.Event event) {
+        actualMatchCount[0]++;
+      }
+
+      @Override
+      public void onEvent(List<io.siddhi.core.event.Event> events) {
+
+      }
+    };
+
+    TrendProcessor trend = new TrendProcessor(callback);
+    trend.onPipelineStarted(processorParams, null, null);
+
+    sendEvents(trend, eventSettings);
+    LOG.info("Expected match count is {}", expectedMatchCount);
+    LOG.info("Actual match count is {}", actualMatchCount[0]);
+    Assertions.assertEquals(expectedMatchCount, actualMatchCount[0]);
+  }
+
+  private void sendEvents(TrendProcessor trend,
+                          List<Tuple2<Integer, Integer>> eventSettings) {
+    List<Tuple2<Integer, Event>> events = makeEvents(eventSettings);
+    for (Tuple2<Integer, Event> event : events) {
+      LOG.info("Sending event with value " + event.v.getFieldBySelector("s0::randomValue"));
+      trend.onEvent(event.v, null);
+      try {
+        Thread.sleep(event.k);
+      } catch (InterruptedException e) {
+        e.printStackTrace();
+      }
+    }
+  }
+
+  private List<Tuple2<Integer, Event>> makeEvents(List<Tuple2<Integer, Integer>> eventSettings) {
+    List<Tuple2<Integer, Event>> events = new ArrayList<>();
+    for (Tuple2<Integer, Integer> eventSetting : eventSettings) {
+      events.add(makeEvent(eventSetting.k, eventSetting.v));
+    }
+    return events;
+  }
+
+  private Tuple2<Integer, Event> makeEvent(Integer timeout, Integer value) {
+    Map<String, Object> map = new HashMap<>();
+    map.put("randomValue", value);
+    return new Tuple2<>(timeout, EventFactory.fromMap(map, new SourceInfo("test"
+            + "-topic", "s0"),
+        new SchemaInfo(null,
+            new ArrayList<>())));
+  }
 }