Repository: apache/streampipes
Pull request #3166 (2024-08-20 16:21:08)

chore(#3165): Remove support for configurable data formats
<!--
  ~ 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/community/get-involved/
    - 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.
-->

Remove for configurable data formats and stick to JSON as the default format for internal messaging.

### 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): yes
* Custom pipeline elements should be updated by removing the registration of data formats in the SDK's builder methods.



Patch:
diff --git a/archetypes/streampipes-archetype-extensions-jvm/src/main/resources/archetype-resources/src/main/java/Init.java b/archetypes/streampipes-archetype-extensions-jvm/src/main/resources/archetype-resources/src/main/java/Init.java
index b467203177..47e910e837 100644
--- a/archetypes/streampipes-archetype-extensions-jvm/src/main/resources/archetype-resources/src/main/java/Init.java
+++ b/archetypes/streampipes-archetype-extensions-jvm/src/main/resources/archetype-resources/src/main/java/Init.java
@@ -21,10 +21,6 @@
 #set( $symbol_escape = '\' )
 package ${package};
 
-import org.apache.streampipes.dataformat.cbor.CborDataFormatFactory;
-import org.apache.streampipes.dataformat.fst.FstDataFormatFactory;
-import org.apache.streampipes.dataformat.json.JsonDataFormatFactory;
-import org.apache.streampipes.dataformat.smile.SmileDataFormatFactory;
 import org.apache.streampipes.extensions.management.model.SpServiceDefinition;
 import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBuilder;
 import org.apache.streampipes.messaging.jms.SpJmsProtocolFactory;
@@ -56,11 +52,6 @@ public class Init extends StreamPipesExtensionsServiceBase {
         .registerPipelineElement(new ${classNamePrefix}DataSink())
         .registerAdapter(new ${classNamePrefix}GenericAdapter())
         .registerAdapter(new ${classNamePrefix}SpecificAdapter())
-        .registerMessagingFormats(
-            new JsonDataFormatFactory(),
-            new CborDataFormatFactory(),
-            new SmileDataFormatFactory(),
-            new FstDataFormatFactory())
         .registerMessagingProtocols(
             new SpKafkaProtocolFactory(),
             new SpJmsProtocolFactory(),
diff --git a/archetypes/streampipes-archetype-pe-processors-flink/src/main/resources/archetype-resources/src/main/java/Init.java b/archetypes/streampipes-archetype-pe-processors-flink/src/main/resources/archetype-resources/src/main/java/Init.java
index 71e371dc77..1a53b264a8 100644
--- a/archetypes/streampipes-archetype-pe-processors-flink/src/main/resources/archetype-resources/src/main/java/Init.java
+++ b/archetypes/streampipes-archetype-pe-processors-flink/src/main/resources/archetype-resources/src/main/java/Init.java
@@ -25,10 +25,6 @@ import ${package}.config.ConfigKeys;
 
 import ${package}.pe.processor.${packageName}.${classNamePrefix}Controller;
 
-import org.apache.streampipes.dataformat.cbor.CborDataFormatFactory;
-import org.apache.streampipes.dataformat.fst.FstDataFormatFactory;
-import org.apache.streampipes.dataformat.json.JsonDataFormatFactory;
-import org.apache.streampipes.dataformat.smile.SmileDataFormatFactory;
 import org.apache.streampipes.extensions.management.model.SpServiceDefinition;
 import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBuilder;
 import org.apache.streampipes.messaging.jms.SpJmsProtocolFactory;
@@ -51,11 +47,6 @@ public class Init extends StreamPipesExtensionsServiceBase {
               "",
               8090)
 	  .registerPipelineElement(new ${classNamePrefix}Controller())
-      .registerMessagingFormats(
-              new JsonDataFormatFactory(),
-              new CborDataFormatFactory(),
-              new SmileDataFormatFactory(),
-              new FstDataFormatFactory())
       .registerMessagingProtocols(
               new SpKafkaProtocolFactory(),
               new SpJmsProtocolFactory(),
diff --git a/archetypes/streampipes-archetype-pe-sinks-flink/src/main/resources/archetype-resources/src/main/java/Init.java b/archetypes/streampipes-archetype-pe-sinks-flink/src/main/resources/archetype-resources/src/main/java/Init.java
index b320c7a3ca..32f521f419 100644
--- a/archetypes/streampipes-archetype-pe-sinks-flink/src/main/resources/archetype-resources/src/main/java/Init.java
+++ b/archetypes/streampipes-archetype-pe-sinks-flink/src/main/resources/archetype-resources/src/main/java/Init.java
@@ -21,10 +21,6 @@
 #set( $symbol_escape = '\' )
 package ${package};
 
-import org.apache.streampipes.dataformat.cbor.CborDataFormatFactory;
-import org.apache.streampipes.dataformat.fst.FstDataFormatFactory;
-import org.apache.streampipes.dataformat.json.JsonDataFormatFactory;
-import org.apache.streampipes.dataformat.smile.SmileDataFormatFactory;
 import org.apache.streampipes.extensions.management.model.SpServiceDefinition;
 import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBuilder;
 import org.apache.streampipes.messaging.jms.SpJmsProtocolFactory;
@@ -50,11 +46,6 @@ public class Init extends StreamPipesExtensionsServiceBase {
             "",
             8090)
         .registerPipelineElement(new ${classNamePrefix}Controller())
-      .registerMessagingFormats(
-        new JsonDataFormatFactory(),
-        new CborDataFormatFactory(),
-        new SmileDataFormatFactory(),
-        new FstDataFormatFactory())
         .registerMessagingProtocols(
             new SpKafkaProtocolFactory(),
             new SpJmsProtocolFactory(),
diff --git a/pom.xml b/pom.xml
index 7b63eb5497..6bcdad2b73 100644
--- a/pom.xml
+++ b/pom.xml
@@ -57,7 +57,6 @@
         <file-management.version>3.1.0</file-management.version>
         <flink.version>1.13.5</flink.version>
         <fogsy-qudt.version>1.0</fogsy-qudt.version>
-        <fst.version>2.57</fst.version>
         <geojson-jackson.version>1.14</geojson-jackson.version>
         <google-maps-services.version>2.2.0</google-maps-services.version>
         <graalvm.js.version>23.0.0</graalvm.js.version>
@@ -187,16 +186,6 @@
                 <artifactId>jackson-databind</artifactId>
                 <version>${jackson.databind.version}</version>
             </dependency>
-            <dependency>
-                <groupId>com.fasterxml.jackson.dataformat</groupId>
-                <artifactId>jackson-dataformat-cbor</artifactId>
-                <version>${jackson.version}</version>
-            </dependency>
-            <dependency>
-                <groupId>com.fasterxml.jackson.dataformat</groupId>
-                <artifactId>jackson-dataformat-smile</artifactId>
-                <version>${jackson.version}</version>
-            </dependency>
             <dependency>
                 <groupId>com.fasterxml.jackson.dataformat</groupId>
                 <artifactId>jackson-dataformat-xml</artifactId>
@@ -242,11 +231,6 @@
                 <artifactId>geojson-jackson</artifactId>
                 <version>${geojson-jackson.version}</version>
             </dependency>
-            <dependency>
-                <groupId>de.ruedigermoeller</groupId>
-                <artifactId>fst</artifactId>
-                <version>${fst.version}</version>
-            </dependency>
             <dependency>
                 <groupId>io.fogsy</groupId>
                 <artifactId>qudt</artifactId>
@@ -827,10 +811,6 @@
         <module>streampipes-data-explorer-management</module>
         <module>streampipes-data-export</module>
         <module>streampipes-dataformat</module>
-        <module>streampipes-dataformat-cbor</module>
-        <module>streampipes-dataformat-fst</module>
-        <module>streampipes-dataformat-json</module>
-        <module>streampipes-dataformat-smile</module>
         <module>streampipes-extensions</module>
         <module>streampipes-integration-tests</module>
         <module>streampipes-mail</module>
diff --git a/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IStreamPipesClient.java b/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IStreamPipesClient.java
index b326b362b9..8e91c1ce45 100644
--- a/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IStreamPipesClient.java
+++ b/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/IStreamPipesClient.java
@@ -28,6 +28,8 @@ import org.apache.streampipes.model.mail.SpEmail;
 import java.io.Serializable;
 
 public interface IStreamPipesClient extends Serializable {
+
+  @Deprecated(since = "0.97.0", forRemoval = true)
   void registerDataFormat(SpDataFormatFactory spDataFormatFactory);
 
   void registerProtocol(SpProtocolDefinitionFactory<?> spProtocolDefinitionFactory);
diff --git a/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/config/IStreamPipesClientConfig.java b/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/config/IStreamPipesClientConfig.java
index 4238eec3d7..5fef7a6ba4 100644
--- a/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/config/IStreamPipesClientConfig.java
+++ b/streampipes-client-api/src/main/java/org/apache/streampipes/client/api/config/IStreamPipesClientConfig.java
@@ -18,7 +18,6 @@
 
 package org.apache.streampipes.client.api.config;
 
-import org.apache.streampipes.dataformat.SpDataFormatFactory;
 import org.apache.streampipes.messaging.SpProtocolDefinitionFactory;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -26,8 +25,6 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 public interface IStreamPipesClientConfig {
   ObjectMapper getSerializer();
 
-  void addDataFormat(SpDataFormatFactory spDataFormatFactory);
-
   void addTransportProtocol(SpProtocolDefinitionFactory<?> protocolDefinitionFactory);
 
   ClientConnectionUrlResolver getConnectionConfig();
diff --git a/streampipes-client/pom.xml b/streampipes-client/pom.xml
index 3707a3e700..af5127564e 100644
--- a/streampipes-client/pom.xml
+++ b/streampipes-client/pom.xml
@@ -36,17 +36,7 @@
         </dependency>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-dataformat-json</artifactId>
-            <version>0.97.0-SNAPSHOT</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-dataformat-cbor</artifactId>
-            <version>0.97.0-SNAPSHOT</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-dataformat-fst</artifactId>
+            <artifactId>streampipes-dataformat</artifactId>
             <version>0.97.0-SNAPSHOT</version>
         </dependency>
         <dependency>
diff --git a/streampipes-client/src/main/java/org/apache/streampipes/client/StreamPipesClient.java b/streampipes-client/src/main/java/org/apache/streampipes/client/StreamPipesClient.java
index 1eee9f475d..a8dfa12b14 100644
--- a/streampipes-client/src/main/java/org/apache/streampipes/client/StreamPipesClient.java
+++ b/streampipes-client/src/main/java/org/apache/streampipes/client/StreamPipesClient.java
@@ -37,9 +37,6 @@ import org.apache.streampipes.client.model.StreamPipesClientConfig;
 import org.apache.streampipes.client.model.StreamPipesClientConnectionConfig;
 import org.apache.streampipes.client.paths.ApiPath;
 import org.apache.streampipes.dataformat.SpDataFormatFactory;
-import org.apache.streampipes.dataformat.cbor.CborDataFormatFactory;
-import org.apache.streampipes.dataformat.fst.FstDataFormatFactory;
-import org.apache.streampipes.dataformat.json.JsonDataFormatFactory;
 import org.apache.streampipes.messaging.SpProtocolDefinitionFactory;
 import org.apache.streampipes.model.mail.SpEmail;
 
@@ -52,9 +49,6 @@ public class StreamPipesClient implements
 
   private StreamPipesClient(ClientConnectionUrlResolver connectionConfig) {
     this.config = new StreamPipesClientConfig(connectionConfig);
-    this.registerDataFormat(new JsonDataFormatFactory());
-    this.registerDataFormat(new FstDataFormatFactory());
-    this.registerDataFormat(new CborDataFormatFactory());
   }
 
   private StreamPipesClient(String streamPipesHost,
@@ -128,11 +122,12 @@ public class StreamPipesClient implements
   /**
    * Register a new data format that is used by the live API
    *
+   * @deprecated
    * @param spDataFormatFactory The data format factory
    */
+  @Deprecated(forRemoval = true, since = "0.97.0")
   @Override
   public void registerDataFormat(SpDataFormatFactory spDataFormatFactory) {
-    this.config.addDataFormat(spDataFormatFactory);
   }
 
   @Override
diff --git a/streampipes-client/src/main/java/org/apache/streampipes/client/live/ProducerManager.java b/streampipes-client/src/main/java/org/apache/streampipes/client/live/ProducerManager.java
index 6cf97026d1..1452d6cfd1 100644
--- a/streampipes-client/src/main/java/org/apache/streampipes/client/live/ProducerManager.java
+++ b/streampipes-client/src/main/java/org/apache/streampipes/client/live/ProducerManager.java
@@ -19,7 +19,6 @@
 package org.apache.streampipes.client.live;
 
 import org.apache.streampipes.client.api.live.IConfiguredEventProducer;
-import org.apache.streampipes.dataformat.SpDataFormatDefinition;
 import org.apache.streampipes.dataformat.SpDataFormatManager;
 import org.apache.streampipes.messaging.EventProducer;
 import org.apache.streampipes.messaging.SpProtocolManager;
@@ -39,7 +38,7 @@ public class ProducerManager {
 
     return new ConfiguredEventProducer(
         producer,
-        findFormatDefinition()
+        SpDataFormatManager.getFormatDefinition()
     );
   }
 
@@ -51,12 +50,4 @@ public class ProducerManager {
         .orElseThrow()
         .getProducer(protocol);
   }
-
-  private SpDataFormatDefinition findFormatDefinition() {
-    var format = grounding.getTransportFormats().get(0);
-    return SpDataFormatManager
-        .INSTANCE
-        .findDefinition(format)
-        .orElseThrow();
-  }
 }
diff --git a/streampipes-client/src/main/java/org/apache/streampipes/client/live/SubscriptionManager.java b/streampipes-client/src/main/java/org/apache/streampipes/client/live/SubscriptionManager.java
index 961d9d3006..947e190f60 100644
--- a/streampipes-client/src/main/java/org/apache/streampipes/client/live/SubscriptionManager.java
+++ b/streampipes-client/src/main/java/org/apache/streampipes/client/live/SubscriptionManager.java
@@ -57,40 +57,31 @@ public class SubscriptionManager {
   }
 
   public ISubscription subscribe() {
-    var formatDefinitionOpt = SpDataFormatManager
-        .INSTANCE
-        .findDefinition(this.grounding.getTransportFormats().get(0));
 
     try {
       SpProtocolDefinition<TransportProtocol> protocolDefinition = findProtocol(getTransportProtocol());
+      final SpDataFormatDefinition converter = SpDataFormatManager.getFormatDefinition();
 
-      if (formatDefinitionOpt.isPresent()) {
-        final SpDataFormatDefinition converter = formatDefinitionOpt.get();
-
-        var protocol = getTransportProtocol();
-        if (overrideSettings) {
-          if (protocol instanceof KafkaTransportProtocol) {
-            brokerConfigOverride.overrideKafkaHostname((KafkaTransportProtocol) protocol);
-          }
-          brokerConfigOverride.overrideHostname(protocol);
-          brokerConfigOverride.overridePort(protocol);
+      var protocol = getTransportProtocol();
+      if (overrideSettings) {
+        if (protocol instanceof KafkaTransportProtocol) {
+          brokerConfigOverride.overrideKafkaHostname((KafkaTransportProtocol) protocol);
         }
+        brokerConfigOverride.overrideHostname(protocol);
+        brokerConfigOverride.overridePort(protocol);
+      }
 
-        EventConsumer consumer = protocolDefinition.getConsumer(protocol);
-        consumer.connect(event -> {
-          try {
-            Event spEvent = EventFactory.fromMap(converter.toMap(event));
-            callback.onEvent(spEvent);
-          } catch (SpRuntimeException e) {
-            e.printStackTrace();
-          }
-        });
+      EventConsumer consumer = protocolDefinition.getConsumer(protocol);
+      consumer.connect(event -> {
+        try {
+          Event spEvent = EventFactory.fromMap(converter.toMap(event));
+          callback.onEvent(spEvent);
+        } catch (SpRuntimeException e) {
+          e.printStackTrace();
+        }
+      });
 
-        return new Subscription(consumer);
-      } else {
-        throw new SpRuntimeException(
-            "No converter found for data format - did you add a format factory (client.registerDataFormat)?");
-      }
+      return new Subscription(consumer);
     } catch (NoSuchElementException e) {
       throw new SpRuntimeException(
           "Could not find an implementation for messaging protocol "
diff --git a/streampipes-client/src/main/java/org/apache/streampipes/client/model/StreamPipesClientConfig.java b/streampipes-client/src/main/java/org/apache/streampipes/client/model/StreamPipesClientConfig.java
index 716ed50b21..88e22d56b5 100644
--- a/streampipes-client/src/main/java/org/apache/streampipes/client/model/StreamPipesClientConfig.java
+++ b/streampipes-client/src/main/java/org/apache/streampipes/client/model/StreamPipesClientConfig.java
@@ -19,8 +19,6 @@ package org.apache.streampipes.client.model;
 
 import org.apache.streampipes.client.api.config.ClientConnectionUrlResolver;
 import org.apache.streampipes.client.api.config.IStreamPipesClientConfig;
-import org.apache.streampipes.dataformat.SpDataFormatFactory;
-import org.apache.streampipes.dataformat.SpDataFormatManager;
 import org.apache.streampipes.messaging.SpProtocolDefinitionFactory;
 import org.apache.streampipes.messaging.SpProtocolManager;
 import org.apache.streampipes.serializers.json.JacksonSerializer;
@@ -42,11 +40,6 @@ public class StreamPipesClientConfig implements IStreamPipesClientConfig {
     return serializer;
   }
 
-  @Override
-  public void addDataFormat(SpDataFormatFactory spDataFormatFactory) {
-    SpDataFormatManager.INSTANCE.register(spDataFormatFactory);
-  }
-
   @Override
   public void addTransportProtocol(SpProtocolDefinitionFactory<?> protocolDefinitionFactory) {
     SpProtocolManager.INSTANCE.register(protocolDefinitionFactory);
diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/GroundingUtils.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/GroundingUtils.java
index 74c771358e..92870f490c 100644
--- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/GroundingUtils.java
+++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/GroundingUtils.java
@@ -29,7 +29,6 @@ import org.apache.streampipes.model.grounding.SimpleTopicDefinition;
 import org.apache.streampipes.model.grounding.TopicDefinition;
 import org.apache.streampipes.model.grounding.TransportProtocol;
 
-import java.util.Collections;
 import java.util.UUID;
 
 public class GroundingUtils {
@@ -82,9 +81,6 @@ public class GroundingUtils {
       );
     }
 
-    eventGrounding.setTransportFormats(Collections
-        .singletonList(TransportFormatGenerator.getTransportFormat()));
-
     return eventGrounding;
   }
 
diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/TransportFormatGenerator.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/TransportFormatGenerator.java
deleted file mode 100644
index 7038866af5..0000000000
--- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/TransportFormatGenerator.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.connect.management.util;
-
-import org.apache.streampipes.model.configuration.SpDataFormat;
-import org.apache.streampipes.model.grounding.TransportFormat;
-import org.apache.streampipes.sdk.helpers.SupportedFormats;
-import org.apache.streampipes.vocabulary.MessageFormat;
-
-import java.util.Arrays;
-import java.util.List;
-
-public class TransportFormatGenerator {
-
-  public static TransportFormat getTransportFormat() {
-    var cfg = Utils.getCoreConfigStorage().get();
-    List<SpDataFormat> supportedFormats =
-        cfg.getMessagingSettings().getPrioritizedFormats();
-
-    if (supportedFormats.size() > 0) {
-      return new TransportFormat(supportedFormats.get(0).getMessageFormat());
-    } else {
-      return new TransportFormat(MessageFormat.JSON);
-    }
-  }
-
-  public static List<TransportFormat> getAllFormats() {
-    return Arrays.asList(SupportedFormats.cborFormat(),
-        SupportedFormats.jsonFormat(),
-        SupportedFormats.fstFormat(),
-        SupportedFormats.smileFormat());
-  }
-}
diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInflux.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInflux.java
index 8c2f674a99..fa35cbe9b9 100644
--- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInflux.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInflux.java
@@ -45,7 +45,7 @@ public class TimeSeriesStorageInflux extends TimeSeriesStorage {
       InfluxClientProvider influxClientProvider
   ) throws SpRuntimeException {
     super(measure);
-    influxDb = influxClientProvider.getInitializedInfluxDBClient(environment);
+    this.influxDb = influxClientProvider.getSetUpInfluxDBClient(environment);
     propertyHandler = new PropertyHandler();
   }
 
diff --git a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/client/InfluxClientProvider.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/client/InfluxClientProvider.java
index f1b47285b4..9f23cee0a6 100644
--- a/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/client/InfluxClientProvider.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/client/InfluxClientProvider.java
@@ -34,8 +34,50 @@ import java.util.concurrent.TimeUnit;
 
 public class InfluxClientProvider {
 
+  private static final int DEFAULT_BATCH_SIZE = 2000;
+  private static final int DEFAULT_FLUSH_DURATION = 500;
+
   private static final Logger LOG = LoggerFactory.getLogger(InfluxClientProvider.class);
 
+  /**
+   * Create a new InfluxDB client from Environment and ensures database is available
+   * @param environment Environment
+   * @return InfluxDB
+   */
+  public InfluxDB getSetUpInfluxDBClient(Environment environment){
+    return getSetUpInfluxDBClient(InfluxConnectionSettings.from(environment));
+  }
+
+  /**
+   * Create a new InfluxDB client from Connection Settings and ensures database is available
+   * @param settings Connection Settings
+   * @return InfluxDB
+   */
+  public InfluxDB getSetUpInfluxDBClient(InfluxConnectionSettings settings){
+    var influxDb = getInitializedInfluxDBClient(settings);
+    this.setupDatabaseAndBatching(influxDb, settings.getDatabaseName());
+
+    return influxDb;
+  }
+
+  /**
+   * Create a new InfluxDB client from provided settings and verify it's available
+   * @param settings Connection settings
+   * @return InfluxDB
+   */
+  public InfluxDB getInitializedInfluxDBClient(InfluxConnectionSettings settings){
+    var influxDb = InfluxClientProvider.getInfluxDBClient(settings);
+
+    // Checking, if server is available
+    var response = influxDb.ping();
+    if (response.getVersion()
+        .equalsIgnoreCase("unknown")) {
+      throw new SpRuntimeException("Could not connect to InfluxDb Server: " + settings.getConnectionUrl());
+    }
+
+    return influxDb;
+  }
+
   /**
    * Create a new InfluxDB client from environment variables
    *
@@ -68,20 +110,23 @@ public class InfluxClientProvider {
     }
   }
 
+  /**
+   * Creates the specified database in the influxDb instance if it does not exist. Enables batching with default values
+   * @param influxDb The InfluxDB client instance
+   * @param databaseName The name of the database
+   */
+  public void setupDatabaseAndBatching(InfluxDB influxDb, String databaseName) {
+    this.setupDatabaseAndBatching(influxDb, databaseName, DEFAULT_BATCH_SIZE, DEFAULT_FLUSH_DURATION);
+  }
 
-  public InfluxDB getInitializedInfluxDBClient(Environment environment) {
-
-    var settings = InfluxConnectionSettings.from(environment);
-    var influxDb = InfluxClientProvider.getInfluxDBClient(settings);
-    var databaseName = settings.getDatabaseName();
-
-    // Checking, if server is available
-    var response = influxDb.ping();
-    if (response.getVersion()
-                .equalsIgnoreCase("unknown")) {
-      throw new SpRuntimeException("Could not connect to InfluxDb Server: " + settings.getConnectionUrl());
-    }
-
+  /**
+   * Creates the specified database in the influxDb instance if it does not exist. Enables batching
+   * @param influxDb The InfluxDB client instance
+   * @param databaseName The name of the database
+   * @param batchSize Batch Size
+   * @param flushDuration Flush Duration
+   */
+  public void setupDatabaseAndBatching(InfluxDB influxDb, String databaseName, int batchSize, int flushDuration) {
     // Checking whether the database exists
     if (!databaseExists(influxDb, databaseName)) {
       LOG.info("Database '" + databaseName + "' not found. Gets created ...");
@@ -90,11 +135,7 @@ public class InfluxClientProvider {
 
     // setting up the database
     influxDb.setDatabase(databaseName);
-    var batchSize = 2000;
-    var flushDuration = 500;
     influxDb.enableBatch(batchSize, flushDuration, TimeUnit.MILLISECONDS);
-
-    return influxDb;
   }
 
   /**
diff --git a/streampipes-dataformat-cbor/pom.xml b/streampipes-dataformat-cbor/pom.xml
deleted file mode 100644
index 9193047547..0000000000
--- a/streampipes-dataformat-cbor/pom.xml
+++ /dev/null
@@ -1,59 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-  ~ 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.
-  ~
-  -->
-
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <parent>
-        <artifactId>streampipes-parent</artifactId>
-        <groupId>org.apache.streampipes</groupId>
-        <version>0.97.0-SNAPSHOT</version>
-    </parent>
-    <modelVersion>4.0.0</modelVersion>
-
-    <artifactId>streampipes-dataformat-cbor</artifactId>
-
-    <dependencies>
-        <!-- StreamPipes dependencies -->
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-dataformat</artifactId>
-            <version>0.97.0-SNAPSHOT</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-vocabulary</artifactId>
-            <version>0.97.0-SNAPSHOT</version>
-        </dependency>
-
-        <!-- External dependencies -->
-        <dependency>
-            <groupId>com.fasterxml.jackson.dataformat</groupId>
-            <artifactId>jackson-dataformat-cbor</artifactId>
-        </dependency>
-    </dependencies>
-
-    <build>
-        <plugins>
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-checkstyle-plugin</artifactId>
-            </plugin>
-        </plugins>
-    </build>
-
-</project>
diff --git a/streampipes-dataformat-cbor/src/main/java/org/apache/streampipes/dataformat/cbor/CborDataFormatDefinition.java b/streampipes-dataformat-cbor/src/main/java/org/apache/streampipes/dataformat/cbor/CborDataFormatDefinition.java
deleted file mode 100644
index 95464b59e8..0000000000
--- a/streampipes-dataformat-cbor/src/main/java/org/apache/streampipes/dataformat/cbor/CborDataFormatDefinition.java
+++ /dev/null
@@ -1,56 +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.dataformat.cbor;
-
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataformat.SpDataFormatDefinition;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.dataformat.cbor.CBORFactory;
-
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-
-public class CborDataFormatDefinition implements SpDataFormatDefinition {
-
-  private ObjectMapper objectMapper;
-
-  public CborDataFormatDefinition() {
-    this.objectMapper = new ObjectMapper(new CBORFactory());
-  }
-
-  @Override
-  public Map<String, Object> toMap(byte[] event) throws SpRuntimeException {
-    try {
-      return objectMapper.readValue(event, HashMap.class);
-    } catch (IOException e) {
-      throw new SpRuntimeException("Could not convert event to map data structure");
-    }
-  }
-
-  @Override
-  public byte[] fromMap(Map<String, Object> event) throws SpRuntimeException {
-    try {
-      return objectMapper.writeValueAsBytes(event);
-    } catch (JsonProcessingException e) {
-      throw new SpRuntimeException("Could not convert map data structure to JSON string");
-    }
-  }
-}
diff --git a/streampipes-dataformat-cbor/src/main/java/org/apache/streampipes/dataformat/cbor/CborDataFormatFactory.java b/streampipes-dataformat-cbor/src/main/java/org/apache/streampipes/dataformat/cbor/CborDataFormatFactory.java
deleted file mode 100644
index 5814a4fb5c..0000000000
--- a/streampipes-dataformat-cbor/src/main/java/org/apache/streampipes/dataformat/cbor/CborDataFormatFactory.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.dataformat.cbor;
-
-import org.apache.streampipes.dataformat.SpDataFormatDefinition;
-import org.apache.streampipes.dataformat.SpDataFormatFactory;
-import org.apache.streampipes.vocabulary.MessageFormat;
-
-public class CborDataFormatFactory extends SpDataFormatFactory {
-
-  @Override
-  public String getTransportFormatRdfUri() {
-    return MessageFormat.CBOR;
-  }
-
-  @Override
-  public SpDataFormatDefinition createInstance() {
-    return new CborDataFormatDefinition();
-  }
-}
diff --git a/streampipes-dataformat-fst/pom.xml b/streampipes-dataformat-fst/pom.xml
deleted file mode 100644
index 951a89e185..0000000000
--- a/streampipes-dataformat-fst/pom.xml
+++ /dev/null
@@ -1,59 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-  ~ 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.
-  ~
-  -->
-
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <parent>
-        <artifactId>streampipes-parent</artifactId>
-        <groupId>org.apache.streampipes</groupId>
-        <version>0.97.0-SNAPSHOT</version>
-    </parent>
-    <modelVersion>4.0.0</modelVersion>
-
-    <artifactId>streampipes-dataformat-fst</artifactId>
-
-    <dependencies>
-        <!-- StreamPipes dependencies -->
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-dataformat</artifactId>
-            <version>0.97.0-SNAPSHOT</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-vocabulary</artifactId>
-            <version>0.97.0-SNAPSHOT</version>
-        </dependency>
-
-        <!-- External dependencies -->
-        <dependency>
-            <groupId>de.ruedigermoeller</groupId>
-            <artifactId>fst</artifactId>
-        </dependency>
-    </dependencies>
-
-    <build>
-        <plugins>
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-checkstyle-plugin</artifactId>
-            </plugin>
-        </plugins>
-    </build>
-
-</project>
diff --git a/streampipes-dataformat-fst/src/main/java/org/apache/streampipes/dataformat/fst/FstDataFormatDefinition.java b/streampipes-dataformat-fst/src/main/java/org/apache/streampipes/dataformat/fst/FstDataFormatDefinition.java
deleted file mode 100644
index 39095143fc..0000000000
--- a/streampipes-dataformat-fst/src/main/java/org/apache/streampipes/dataformat/fst/FstDataFormatDefinition.java
+++ /dev/null
@@ -1,44 +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.dataformat.fst;
-
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataformat.SpDataFormatDefinition;
-
-import org.nustaq.serialization.FSTConfiguration;
-
-import java.util.Map;
-
-public class FstDataFormatDefinition implements SpDataFormatDefinition {
-
-  private static FSTConfiguration conf = FSTConfiguration.createDefaultConfiguration();
-
-  public FstDataFormatDefinition() {
-
-  }
-
-  @Override
-  public Map<String, Object> toMap(byte[] event) throws SpRuntimeException {
-    return (Map<String, Object>) conf.asObject(event);
-  }
-
-  @Override
-  public byte[] fromMap(Map<String, Object> event) throws SpRuntimeException {
-    return conf.asByteArray(event);
-  }
-}
diff --git a/streampipes-dataformat-fst/src/main/java/org/apache/streampipes/dataformat/fst/FstDataFormatFactory.java b/streampipes-dataformat-fst/src/main/java/org/apache/streampipes/dataformat/fst/FstDataFormatFactory.java
deleted file mode 100644
index 285ea9cc25..0000000000
--- a/streampipes-dataformat-fst/src/main/java/org/apache/streampipes/dataformat/fst/FstDataFormatFactory.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.dataformat.fst;
-
-import org.apache.streampipes.dataformat.SpDataFormatDefinition;
-import org.apache.streampipes.dataformat.SpDataFormatFactory;
-import org.apache.streampipes.vocabulary.MessageFormat;
-
-public class FstDataFormatFactory extends SpDataFormatFactory {
-
-  @Override
-  public String getTransportFormatRdfUri() {
-    return MessageFormat.FST;
-  }
-
-  @Override
-  public SpDataFormatDefinition createInstance() {
-    return new FstDataFormatDefinition();
-  }
-}
diff --git a/streampipes-dataformat-json/pom.xml b/streampipes-dataformat-json/pom.xml
deleted file mode 100644
index 493a3197d9..0000000000
--- a/streampipes-dataformat-json/pom.xml
+++ /dev/null
@@ -1,63 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-  ~ 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.
-  ~
-  -->
-
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <parent>
-        <artifactId>streampipes-parent</artifactId>
-        <groupId>org.apache.streampipes</groupId>
-        <version>0.97.0-SNAPSHOT</version>
-    </parent>
-    <modelVersion>4.0.0</modelVersion>
-
-    <artifactId>streampipes-dataformat-json</artifactId>
-
-    <dependencies>
-        <!-- StreamPipes dependencies -->
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-dataformat</artifactId>
-            <version>0.97.0-SNAPSHOT</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-vocabulary</artifactId>
-            <version>0.97.0-SNAPSHOT</version>
-        </dependency>
-
-        <!-- External dependencies -->
-        <dependency>
-            <groupId>com.fasterxml.jackson.core</groupId>
-            <artifactId>jackson-databind</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>com.fasterxml.jackson.core</groupId>
-            <artifactId>jackson-core</artifactId>
-        </dependency>
-    </dependencies>
-
-    <build>
-        <plugins>
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-checkstyle-plugin</artifactId>
-            </plugin>
-        </plugins>
-    </build>
-
-</project>
diff --git a/streampipes-dataformat-json/src/main/java/org/apache/streampipes/dataformat/json/JsonDataFormatFactory.java b/streampipes-dataformat-json/src/main/java/org/apache/streampipes/dataformat/json/JsonDataFormatFactory.java
deleted file mode 100644
index 87821cd37c..0000000000
--- a/streampipes-dataformat-json/src/main/java/org/apache/streampipes/dataformat/json/JsonDataFormatFactory.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.dataformat.json;
-
-import org.apache.streampipes.dataformat.SpDataFormatDefinition;
-import org.apache.streampipes.dataformat.SpDataFormatFactory;
-import org.apache.streampipes.vocabulary.MessageFormat;
-
-public class JsonDataFormatFactory extends SpDataFormatFactory {
-
-  @Override
-  public String getTransportFormatRdfUri() {
-    return MessageFormat.JSON;
-  }
-
-  @Override
-  public SpDataFormatDefinition createInstance() {
-    return new JsonDataFormatDefinition();
-  }
-}
diff --git a/streampipes-dataformat-smile/pom.xml b/streampipes-dataformat-smile/pom.xml
deleted file mode 100644
index e8825fd645..0000000000
--- a/streampipes-dataformat-smile/pom.xml
+++ /dev/null
@@ -1,59 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-  ~ 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.
-  ~
-  -->
-
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <parent>
-        <artifactId>streampipes-parent</artifactId>
-        <groupId>org.apache.streampipes</groupId>
-        <version>0.97.0-SNAPSHOT</version>
-    </parent>
-    <modelVersion>4.0.0</modelVersion>
-
-    <artifactId>streampipes-dataformat-smile</artifactId>
-
-    <dependencies>
-        <!-- StreamPipes dependencies -->
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-dataformat</artifactId>
-            <version>0.97.0-SNAPSHOT</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-vocabulary</artifactId>
-            <version>0.97.0-SNAPSHOT</version>
-        </dependency>
-
-        <!-- External dependencies -->
-        <dependency>
-            <groupId>com.fasterxml.jackson.dataformat</groupId>
-            <artifactId>jackson-dataformat-smile</artifactId>
-        </dependency>
-    </dependencies>
-
-    <build>
-        <plugins>
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-checkstyle-plugin</artifactId>
-            </plugin>
-        </plugins>
-    </build>
-
-</project>
diff --git a/streampipes-dataformat-smile/src/main/java/org/apache/streampipes/dataformat/smile/SmileDataFormatDefinition.java b/streampipes-dataformat-smile/src/main/java/org/apache/streampipes/dataformat/smile/SmileDataFormatDefinition.java
deleted file mode 100644
index 6c6bbdf26f..0000000000
--- a/streampipes-dataformat-smile/src/main/java/org/apache/streampipes/dataformat/smile/SmileDataFormatDefinition.java
+++ /dev/null
@@ -1,56 +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.dataformat.smile;
-
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataformat.SpDataFormatDefinition;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.dataformat.smile.SmileFactory;
-
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-
-public class SmileDataFormatDefinition implements SpDataFormatDefinition {
-
-  private ObjectMapper objectMapper;
-
-  public SmileDataFormatDefinition() {
-    this.objectMapper = new ObjectMapper(new SmileFactory());
-  }
-
-  @Override
-  public Map<String, Object> toMap(byte[] event) throws SpRuntimeException {
-    try {
-      return objectMapper.readValue(event, HashMap.class);
-    } catch (IOException e) {
-      throw new SpRuntimeException("Could not convert event to map data structure");
-    }
-  }
-
-  @Override
-  public byte[] fromMap(Map<String, Object> event) throws SpRuntimeException {
-    try {
-      return objectMapper.writeValueAsBytes(event);
-    } catch (JsonProcessingException e) {
-      throw new SpRuntimeException("Could not convert map data structure to JSON string");
-    }
-  }
-}
diff --git a/streampipes-dataformat-smile/src/main/java/org/apache/streampipes/dataformat/smile/SmileDataFormatFactory.java b/streampipes-dataformat-smile/src/main/java/org/apache/streampipes/dataformat/smile/SmileDataFormatFactory.java
deleted file mode 100644
index d04b15197f..0000000000
--- a/streampipes-dataformat-smile/src/main/java/org/apache/streampipes/dataformat/smile/SmileDataFormatFactory.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.dataformat.smile;
-
-import org.apache.streampipes.dataformat.SpDataFormatDefinition;
-import org.apache.streampipes.dataformat.SpDataFormatFactory;
-import org.apache.streampipes.vocabulary.MessageFormat;
-
-public class SmileDataFormatFactory extends SpDataFormatFactory {
-  @Override
-  public String getTransportFormatRdfUri() {
-    return MessageFormat.SMILE;
-  }
-
-  @Override
-  public SpDataFormatDefinition createInstance() {
-    return new SmileDataFormatDefinition();
-  }
-}
diff --git a/streampipes-dataformat/pom.xml b/streampipes-dataformat/pom.xml
index f35330a3d6..daefc05026 100644
--- a/streampipes-dataformat/pom.xml
+++ b/streampipes-dataformat/pom.xml
@@ -34,6 +34,14 @@
             <artifactId>streampipes-model</artifactId>
             <version>0.97.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-core</artifactId>
+        </dependency>
     </dependencies>
 
     <build>
diff --git a/streampipes-dataformat-json/src/main/java/org/apache/streampipes/dataformat/json/JsonDataFormatDefinition.java b/streampipes-dataformat/src/main/java/org/apache/streampipes/dataformat/JsonDataFormatDefinition.java
similarity index 92%
rename from streampipes-dataformat-json/src/main/java/org/apache/streampipes/dataformat/json/JsonDataFormatDefinition.java
rename to streampipes-dataformat/src/main/java/org/apache/streampipes/dataformat/JsonDataFormatDefinition.java
index f3bbacaf06..7433202edb 100644
--- a/streampipes-dataformat-json/src/main/java/org/apache/streampipes/dataformat/json/JsonDataFormatDefinition.java
+++ b/streampipes-dataformat/src/main/java/org/apache/streampipes/dataformat/JsonDataFormatDefinition.java
@@ -16,10 +16,9 @@
  *
  */
 
-package org.apache.streampipes.dataformat.json;
+package org.apache.streampipes.dataformat;
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataformat.SpDataFormatDefinition;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -30,7 +29,7 @@ import java.util.Map;
 
 public class JsonDataFormatDefinition implements SpDataFormatDefinition {
 
-  private ObjectMapper objectMapper;
+  private final ObjectMapper objectMapper;
 
   public JsonDataFormatDefinition() {
     this.objectMapper = new ObjectMapper();
diff --git a/streampipes-dataformat/src/main/java/org/apache/streampipes/dataformat/SpDataFormatFactory.java b/streampipes-dataformat/src/main/java/org/apache/streampipes/dataformat/SpDataFormatFactory.java
index b805a23d0d..ca7d9afc66 100644
--- a/streampipes-dataformat/src/main/java/org/apache/streampipes/dataformat/SpDataFormatFactory.java
+++ b/streampipes-dataformat/src/main/java/org/apache/streampipes/dataformat/SpDataFormatFactory.java
@@ -18,15 +18,6 @@
 
 package org.apache.streampipes.dataformat;
 
-import org.apache.streampipes.model.grounding.TransportFormat;
-
 public abstract class SpDataFormatFactory {
 
-  public TransportFormat getTransportFormat() {
-    return new TransportFormat(getTransportFormatRdfUri());
-  }
-
-  public abstract String getTransportFormatRdfUri();
-
-  public abstract SpDataFormatDefinition createInstance();
 }
diff --git a/streampipes-dataformat/src/main/java/org/apache/streampipes/dataformat/SpDataFormatManager.java b/streampipes-dataformat/src/main/java/org/apache/streampipes/dataformat/SpDataFormatManager.java
index 342a9b40d8..6b8ff4d7ef 100644
--- a/streampipes-dataformat/src/main/java/org/apache/streampipes/dataformat/SpDataFormatManager.java
+++ b/streampipes-dataformat/src/main/java/org/apache/streampipes/dataformat/SpDataFormatManager.java
@@ -18,43 +18,9 @@
 
 package org.apache.streampipes.dataformat;
 
-import org.apache.streampipes.model.grounding.TransportFormat;
+public class SpDataFormatManager {
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Optional;
-
-public enum SpDataFormatManager {
-
-  INSTANCE;
-
-  private List<SpDataFormatFactory> availableDataFormats;
-
-  SpDataFormatManager() {
-    this.availableDataFormats = new ArrayList<>();
-  }
-
-  public void register(SpDataFormatFactory dataFormatDefinition) {
-    availableDataFormats.add(dataFormatDefinition);
+  public static SpDataFormatDefinition getFormatDefinition() {
+    return new JsonDataFormatDefinition();
   }
-
-  public List<SpDataFormatFactory> getAvailableDataFormats() {
-    return availableDataFormats;
-  }
-
-  public Optional<SpDataFormatDefinition> findDefinition(TransportFormat transportFormat) {
-    // TODO why is transportFormat.getRdfType a list?
-    return this.availableDataFormats
-        .stream()
-        .filter
-            (adf -> transportFormat
-                .getRdfType()
-                .stream()
-                .anyMatch(tf -> tf.toString().equals(adf
-                    .getTransportFormatRdfUri())))
-        .map(SpDataFormatFactory::createInstance)
-        .findFirst();
-
-  }
-
 }
diff --git a/streampipes-extensions-management/pom.xml b/streampipes-extensions-management/pom.xml
index d4ca517c8c..3c0655c595 100644
--- a/streampipes-extensions-management/pom.xml
+++ b/streampipes-extensions-management/pom.xml
@@ -48,11 +48,6 @@
             <artifactId>streampipes-dataformat</artifactId>
             <version>0.97.0-SNAPSHOT</version>
         </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-dataformat-smile</artifactId>
-            <version>0.97.0-SNAPSHOT</version>
-        </dependency>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
             <artifactId>streampipes-messaging</artifactId>
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToBrokerAdapterSink.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToBrokerAdapterSink.java
index 66983df357..d63cddc8f1 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToBrokerAdapterSink.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/preprocessing/elements/SendToBrokerAdapterSink.java
@@ -20,15 +20,14 @@ package org.apache.streampipes.extensions.management.connect.adapter.preprocessi
 import org.apache.streampipes.commons.environment.Environment;
 import org.apache.streampipes.commons.environment.Environments;
 import org.apache.streampipes.dataformat.SpDataFormatDefinition;
+import org.apache.streampipes.dataformat.SpDataFormatManager;
 import org.apache.streampipes.extensions.api.connect.IAdapterPipelineElement;
 import org.apache.streampipes.extensions.api.monitoring.SpMonitoringManager;
-import org.apache.streampipes.extensions.management.connect.adapter.util.TransportFormatSelector;
 import org.apache.streampipes.extensions.management.monitoring.ExtensionsLogger;
 import org.apache.streampipes.messaging.EventProducer;
 import org.apache.streampipes.messaging.SpProtocolManager;
 import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 import org.apache.streampipes.model.grounding.KafkaTransportProtocol;
-import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.model.grounding.TransportProtocol;
 
 import java.util.Map;
@@ -54,13 +53,7 @@ public class SendToBrokerAdapterSink implements IAdapterPipelineElement {
     if (producerOpt.isPresent()) {
       this.producer = producerOpt.get().getProducer(this.protocol);
 
-      TransportFormat transportFormat = adapterDescription
-          .getEventGrounding()
-          .getTransportFormats()
-          .get(0);
-
-      this.dataFormatDefinition =
-          new TransportFormatSelector(transportFormat).getDataFormatDefinition();
+      this.dataFormatDefinition = SpDataFormatManager.getFormatDefinition();
 
       producer.connect();
     } else {
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/util/TransportFormatSelector.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/util/TransportFormatSelector.java
deleted file mode 100644
index 40d985c6a6..0000000000
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/connect/adapter/util/TransportFormatSelector.java
+++ /dev/null
@@ -1,75 +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.extensions.management.connect.adapter.util;
-
-import org.apache.streampipes.dataformat.SpDataFormatDefinition;
-import org.apache.streampipes.dataformat.cbor.CborDataFormatDefinition;
-import org.apache.streampipes.dataformat.fst.FstDataFormatDefinition;
-import org.apache.streampipes.dataformat.json.JsonDataFormatDefinition;
-import org.apache.streampipes.dataformat.smile.SmileDataFormatDefinition;
-import org.apache.streampipes.model.grounding.TransportFormat;
-import org.apache.streampipes.vocabulary.MessageFormat;
-
-public class TransportFormatSelector {
-
-  private TransportFormat transportFormat;
-
-  public TransportFormatSelector(TransportFormat format) {
-    this.transportFormat = format;
-  }
-
-  public SpDataFormatDefinition getDataFormatDefinition() {
-    if (isJsonFormat(transportFormat)) {
-      return new JsonDataFormatDefinition();
-    } else if (isCborFormat(transportFormat)) {
-      return new CborDataFormatDefinition();
-    } else if (isFstFormat(transportFormat)) {
-      return new FstDataFormatDefinition();
-    } else if (isSmileFormat(transportFormat)) {
-      return new SmileDataFormatDefinition();
-    } else {
-      throw new IllegalArgumentException("Wrong transport format: " + makeError(transportFormat));
-    }
-  }
-
-  private boolean isSmileFormat(TransportFormat transportFormat) {
-    return isFormat(MessageFormat.SMILE, transportFormat);
-  }
-
-  private boolean isFstFormat(TransportFormat transportFormat) {
-    return isFormat(MessageFormat.FST, transportFormat);
-  }
-
-  private boolean isCborFormat(TransportFormat transportFormat) {
-    return isFormat(MessageFormat.CBOR, transportFormat);
-  }
-
-  private Boolean isJsonFormat(TransportFormat transportFormat) {
-    return isFormat(MessageFormat.JSON, transportFormat);
-  }
-
-  private boolean isFormat(String format, TransportFormat transportFormat) {
-    return transportFormat.getRdfType().stream().anyMatch(tf -> tf.toString().equals(format));
-  }
-
-  private String makeError(TransportFormat transportFormat) {
-    StringBuilder builder = new StringBuilder();
-    transportFormat.getRdfType().forEach(type -> builder.append(type.toString()).append(", "));
-    return builder.toString();
-  }
-}
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/init/DeclarersSingleton.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/init/DeclarersSingleton.java
index fed5a0baa4..311e15ced7 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/init/DeclarersSingleton.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/init/DeclarersSingleton.java
@@ -18,8 +18,6 @@
 
 package org.apache.streampipes.extensions.management.init;
 
-import org.apache.streampipes.dataformat.SpDataFormatFactory;
-import org.apache.streampipes.dataformat.SpDataFormatManager;
 import org.apache.streampipes.extensions.api.connect.StreamPipesAdapter;
 import org.apache.streampipes.extensions.api.declarer.IStreamPipesFunctionDeclarer;
 import org.apache.streampipes.extensions.api.pe.IStreamPipesDataProcessor;
@@ -30,15 +28,10 @@ import org.apache.streampipes.extensions.api.pe.runtime.IStreamPipesRuntimeProvi
 import org.apache.streampipes.extensions.management.model.SpServiceDefinition;
 import org.apache.streampipes.messaging.SpProtocolDefinitionFactory;
 import org.apache.streampipes.messaging.SpProtocolManager;
-import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.model.grounding.TransportProtocol;
 import org.apache.streampipes.model.util.Cloner;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
@@ -48,7 +41,6 @@ import java.util.stream.Collectors;
 
 public class DeclarersSingleton implements IDeclarersSingleton {
 
-  private static final Logger LOG = LoggerFactory.getLogger(DeclarersSingleton.class);
   private static final String Http = "http://";
   private static final String Colon = ":";
   private static final String Slash = "/";
@@ -61,8 +53,6 @@ public class DeclarersSingleton implements IDeclarersSingleton {
   private final Map<String, IStreamPipesFunctionDeclarer> functions;
 
   private final Map<String, TransportProtocol> supportedProtocols;
-  private final Map<String, TransportFormat> supportedFormats;
-
   private final Map<String, StreamPipesAdapter> adapters;
 
   private List<IStreamPipesRuntimeProvider> runtimeProviders;
@@ -80,7 +70,6 @@ public class DeclarersSingleton implements IDeclarersSingleton {
     this.dataSinks = new HashMap<>();
     this.dataStreams = new HashMap<>();
     this.supportedProtocols = new HashMap<>();
-    this.supportedFormats = new HashMap<>();
     this.adapters = new HashMap<>();
     this.functions = new HashMap<>();
     this.runtimeProviders = new ArrayList<>();
@@ -103,7 +92,6 @@ public class DeclarersSingleton implements IDeclarersSingleton {
     this.serviceId = serviceDef.getServiceId();
     this.serviceGroup = serviceDef.getServiceGroup();
     this.registerProtocols(serviceDef.getProtocolDefinitionFactories());
-    this.registerDataFormats(serviceDef.getDataFormatFactories());
     this.runtimeProviders = serviceDef.getRuntimeProviders();
     serviceDef.getAdapters().forEach(a -> this.adapters.put(a.declareConfig().getAdapterDescription().getAppId(), a));
     serviceDef.getFunctions().forEach(f -> this.functions.put(f.getFunctionConfig().getFunctionId().getId(), f));
@@ -134,7 +122,6 @@ public class DeclarersSingleton implements IDeclarersSingleton {
     result.putAll(dataProcessors);
     result.putAll(dataStreams);
     result.putAll(dataSinks);
-    //result.putAll(pipelineTemplateDeclarers);
     return result;
   }
 
@@ -148,20 +135,6 @@ public class DeclarersSingleton implements IDeclarersSingleton {
     protocols.forEach(this::registerProtocol);
   }
 
-  public void registerDataFormat(SpDataFormatFactory dataFormatDefinition) {
-    SpDataFormatManager.INSTANCE.register(dataFormatDefinition);
-    this.supportedFormats.put(dataFormatDefinition.getTransportFormatRdfUri(),
-        dataFormatDefinition.getTransportFormat());
-  }
-
-  public void registerDataFormats(SpDataFormatFactory... dataFormatDefinitions) {
-    registerDataFormats(Arrays.asList(dataFormatDefinitions));
-  }
-
-  public void registerDataFormats(List<SpDataFormatFactory> dataFormatDefinitions) {
-    dataFormatDefinitions.forEach(this::registerDataFormat);
-  }
-
   private void addDataProcessor(IStreamPipesDataProcessor dataProcessor) {
     dataProcessors.put(dataProcessor.declareConfig().getDescription().getAppId(), dataProcessor);
   }
@@ -195,13 +168,6 @@ public class DeclarersSingleton implements IDeclarersSingleton {
         .collect(Collectors.toList());
   }
 
-  public Collection<TransportFormat> getSupportedFormats() {
-    return this.supportedFormats.values()
-        .stream()
-        .map(TransportFormat::new)
-        .collect(Collectors.toList());
-  }
-
   public int getPort() {
     return this.port;
   }
@@ -260,15 +226,6 @@ public class DeclarersSingleton implements IDeclarersSingleton {
     }
   }
 
-  public SpServiceDefinition toServiceDefinition(String serviceId) {
-    SpServiceDefinition serviceDef = new SpServiceDefinition();
-    serviceDef.setServiceId(serviceId);
-    serviceDef.setDefaultPort(this.getPort());
-
-    // TODO create complete service definition
-    return serviceDef;
-  }
-
   public String getServiceGroup() {
     return serviceGroup;
   }
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/model/SpServiceDefinition.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/model/SpServiceDefinition.java
index 3c7f3af96e..42bd2dcc18 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/model/SpServiceDefinition.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/model/SpServiceDefinition.java
@@ -17,7 +17,6 @@
  */
 package org.apache.streampipes.extensions.management.model;
 
-import org.apache.streampipes.dataformat.SpDataFormatFactory;
 import org.apache.streampipes.extensions.api.connect.StreamPipesAdapter;
 import org.apache.streampipes.extensions.api.declarer.IStreamPipesFunctionDeclarer;
 import org.apache.streampipes.extensions.api.migration.IModelMigrator;
@@ -41,7 +40,6 @@ public class SpServiceDefinition {
   private Integer defaultPort;
 
   private List<IStreamPipesPipelineElement<?>> pipelineElements;
-  private List<SpDataFormatFactory> dataFormatFactories;
   private List<SpProtocolDefinitionFactory<?>> protocolDefinitionFactories;
   private List<IStreamPipesFunctionDeclarer> functions;
 
@@ -55,7 +53,6 @@ public class SpServiceDefinition {
   public SpServiceDefinition() {
     this.serviceId = UUID.randomUUID().toString();
     this.pipelineElements = new ArrayList<>();
-    this.dataFormatFactories = new ArrayList<>();
     this.protocolDefinitionFactories = new ArrayList<>();
     this.kvConfigs = new ArrayList<>();
     this.functions = new ArrayList<>();
@@ -120,14 +117,6 @@ public class SpServiceDefinition {
     this.pipelineElements = pipelineElements;
   }
 
-  public void addDataFormatFactory(SpDataFormatFactory factory) {
-    this.dataFormatFactories.add(factory);
-  }
-
-  public void addDataFormatFactories(List<SpDataFormatFactory> factories) {
-    this.dataFormatFactories.addAll(factories);
-  }
-
   public void addConfig(ConfigItem configItem) {
     this.kvConfigs.add(configItem);
   }
@@ -140,14 +129,6 @@ public class SpServiceDefinition {
     this.adapters.addAll(adapters);
   }
 
-  public List<SpDataFormatFactory> getDataFormatFactories() {
-    return dataFormatFactories;
-  }
-
-  public void setDataFormatFactories(List<SpDataFormatFactory> dataFormatFactories) {
-    this.dataFormatFactories = dataFormatFactories;
-  }
-
   public void addProtocolDefinitionFactory(SpProtocolDefinitionFactory<?> factory) {
     this.protocolDefinitionFactories.add(factory);
   }
diff --git a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/model/SpServiceDefinitionBuilder.java b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/model/SpServiceDefinitionBuilder.java
index 1cb2bb004d..ac02d4533b 100644
--- a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/model/SpServiceDefinitionBuilder.java
+++ b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/model/SpServiceDefinitionBuilder.java
@@ -117,13 +117,19 @@ public class SpServiceDefinitionBuilder {
   }
 
 
+  /**
+   * @deprecated data format registration is no longer required
+   */
+  @Deprecated(since = "0.97.0", forRemoval = true)
   public SpServiceDefinitionBuilder registerMessagingFormat(SpDataFormatFactory dataFormatDefinition) {
-    this.serviceDefinition.addDataFormatFactory(dataFormatDefinition);
     return this;
   }
 
+  /**
+   * @deprecated data format registration is no longer required
+   */
+  @Deprecated(since = "0.97.0", forRemoval = true)
   public SpServiceDefinitionBuilder registerMessagingFormats(SpDataFormatFactory... dataFormatDefinitions) {
-    this.serviceDefinition.addDataFormatFactories(Arrays.asList(dataFormatDefinitions));
     return this;
   }
 
diff --git a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/adapter/InfluxDbClient.java b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/adapter/InfluxDbClient.java
index d23af15257..6c83056cd4 100644
--- a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/adapter/InfluxDbClient.java
+++ b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/adapter/InfluxDbClient.java
@@ -109,7 +109,7 @@ public class InfluxDbClient extends SharedInfluxClient {
 
   public void disconnect() {
     if (connected) {
-      influxDb.close();
+      super.disconnect();
       connected = false;
     }
   }
diff --git a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/shared/SharedInfluxClient.java b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/shared/SharedInfluxClient.java
index c455f31aa9..69a3bb50ba 100644
--- a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/shared/SharedInfluxClient.java
+++ b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/shared/SharedInfluxClient.java
@@ -23,7 +23,6 @@ import org.apache.streampipes.dataexplorer.influx.client.InfluxClientProvider;
 import org.apache.streampipes.dataexplorer.influx.client.InfluxConnectionSettings;
 
 import org.influxdb.InfluxDB;
-import org.influxdb.dto.Pong;
 
 public abstract class SharedInfluxClient {
 
@@ -42,12 +41,14 @@ public abstract class SharedInfluxClient {
 
 
   protected void initClient() throws SpRuntimeException {
-    this.influxDb = InfluxClientProvider.getInfluxDBClient(connectionSettings);
+    InfluxClientProvider influxClientProvider = new InfluxClientProvider();
+    this.influxDb = influxClientProvider.getInitializedInfluxDBClient(connectionSettings);
+  }
 
-    // Checking, if server is available
-    Pong response = influxDb.ping();
-    if (response.getVersion().equalsIgnoreCase("unknown")) {
-      throw new SpRuntimeException("Could not connect to InfluxDb Server: " + connectionSettings.getConnectionUrl());
-    }
+  /**
+   * Shuts down the connection to the InfluxDB server
+   */
+  public void disconnect() {
+    influxDb.close();
   }
 }
diff --git a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/sink/InfluxDbClient.java b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/sink/InfluxDbClient.java
index 9c6f49f411..2a33f2cd41 100644
--- a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/sink/InfluxDbClient.java
+++ b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/sink/InfluxDbClient.java
@@ -24,17 +24,13 @@ import org.apache.streampipes.dataexplorer.influx.client.InfluxConnectionSetting
 import org.apache.streampipes.extensions.connectors.influx.shared.SharedInfluxClient;
 import org.apache.streampipes.model.runtime.Event;
 
-import org.influxdb.BatchOptions;
 import org.influxdb.dto.Point;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 public class InfluxDbClient extends SharedInfluxClient {
 
-  private static final Logger LOG = LoggerFactory.getLogger(InfluxDbClient.class);
 
   private final String timestampField;
   private final Integer batchSize;
@@ -66,17 +62,8 @@ public class InfluxDbClient extends SharedInfluxClient {
    */
   private void connect() throws SpRuntimeException {
     super.initClient();
-    var databaseName = connectionSettings.getDatabaseName();
-
-    // Checking whether the database exists
-    if (!influxClientProvider.databaseExists(influxDb, databaseName)) {
-      LOG.info("Database '" + databaseName + "' not found. Gets created ...");
-      influxClientProvider.createDatabase(influxDb, databaseName);
-    }
-
-    // setting up the database
-    influxDb.setDatabase(databaseName);
-    influxDb.enableBatch(BatchOptions.DEFAULTS.actions(batchSize).flushDuration(flushDuration));
+    influxClientProvider.setupDatabaseAndBatching(
+        influxDb, connectionSettings.getDatabaseName(), batchSize, flushDuration);
   }
 
   /**
@@ -107,11 +94,4 @@ public class InfluxDbClient extends SharedInfluxClient {
 
     influxDb.write(p.build());
   }
-
-  /**
-   * Shuts down the connection to the InfluxDB server
-   */
-  void stop() {
-    influxDb.close();
-  }
 }
diff --git a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/sink/InfluxDbSink.java b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/sink/InfluxDbSink.java
index 18f3d2fb54..3da9bc629c 100644
--- a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/sink/InfluxDbSink.java
+++ b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/sink/InfluxDbSink.java
@@ -83,7 +83,7 @@ public class InfluxDbSink extends StreamPipesDataSink {
 
   @Override
   public void onDetach() throws SpRuntimeException {
-    influxDbClient.stop();
+    influxDbClient.disconnect();
   }
 
   public static String prepareString(String s) {
diff --git a/streampipes-extensions/streampipes-connectors-kafka/src/main/java/org/apache/streampipes/extensions/connectors/kafka/sink/KafkaPublishSink.java b/streampipes-extensions/streampipes-connectors-kafka/src/main/java/org/apache/streampipes/extensions/connectors/kafka/sink/KafkaPublishSink.java
index c84a780488..238bbb8048 100644
--- a/streampipes-extensions/streampipes-connectors-kafka/src/main/java/org/apache/streampipes/extensions/connectors/kafka/sink/KafkaPublishSink.java
+++ b/streampipes-extensions/streampipes-connectors-kafka/src/main/java/org/apache/streampipes/extensions/connectors/kafka/sink/KafkaPublishSink.java
@@ -19,7 +19,7 @@
 package org.apache.streampipes.extensions.connectors.kafka.sink;
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataformat.json.JsonDataFormatDefinition;
+import org.apache.streampipes.dataformat.JsonDataFormatDefinition;
 import org.apache.streampipes.extensions.api.pe.IStreamPipesDataSink;
 import org.apache.streampipes.extensions.api.pe.config.IDataSinkConfiguration;
 import org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext;
diff --git a/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/sink/common/MqttClient.java b/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/sink/common/MqttClient.java
index a780653c27..21b5e786df 100644
--- a/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/sink/common/MqttClient.java
+++ b/streampipes-extensions/streampipes-connectors-mqtt/src/main/java/org/apache/streampipes/extensions/connectors/mqtt/sink/common/MqttClient.java
@@ -18,7 +18,7 @@
 package org.apache.streampipes.extensions.connectors.mqtt.sink.common;
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataformat.json.JsonDataFormatDefinition;
+import org.apache.streampipes.dataformat.JsonDataFormatDefinition;
 import org.apache.streampipes.extensions.api.pe.param.IDataSinkParameters;
 import org.apache.streampipes.model.runtime.Event;
 
diff --git a/streampipes-extensions/streampipes-connectors-nats/src/main/java/org/apache/streampipes/extensions/connectors/nats/sink/NatsSink.java b/streampipes-extensions/streampipes-connectors-nats/src/main/java/org/apache/streampipes/extensions/connectors/nats/sink/NatsSink.java
index ceed64e61a..2a2c097063 100644
--- a/streampipes-extensions/streampipes-connectors-nats/src/main/java/org/apache/streampipes/extensions/connectors/nats/sink/NatsSink.java
+++ b/streampipes-extensions/streampipes-connectors-nats/src/main/java/org/apache/streampipes/extensions/connectors/nats/sink/NatsSink.java
@@ -19,7 +19,7 @@
 package org.apache.streampipes.extensions.connectors.nats.sink;
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataformat.json.JsonDataFormatDefinition;
+import org.apache.streampipes.dataformat.JsonDataFormatDefinition;
 import org.apache.streampipes.extensions.api.pe.IStreamPipesDataSink;
 import org.apache.streampipes.extensions.api.pe.config.IDataSinkConfiguration;
 import org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext;
diff --git a/streampipes-extensions/streampipes-connectors-pulsar/src/main/java/org/apache/streampipes/extensions/connectors/pulsar/sink/PulsarPublisherSink.java b/streampipes-extensions/streampipes-connectors-pulsar/src/main/java/org/apache/streampipes/extensions/connectors/pulsar/sink/PulsarPublisherSink.java
index 5be4894b8a..af5d35793a 100644
--- a/streampipes-extensions/streampipes-connectors-pulsar/src/main/java/org/apache/streampipes/extensions/connectors/pulsar/sink/PulsarPublisherSink.java
+++ b/streampipes-extensions/streampipes-connectors-pulsar/src/main/java/org/apache/streampipes/extensions/connectors/pulsar/sink/PulsarPublisherSink.java
@@ -19,7 +19,7 @@ package org.apache.streampipes.extensions.connectors.pulsar.sink;
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
 import org.apache.streampipes.dataformat.SpDataFormatDefinition;
-import org.apache.streampipes.dataformat.json.JsonDataFormatDefinition;
+import org.apache.streampipes.dataformat.JsonDataFormatDefinition;
 import org.apache.streampipes.extensions.api.pe.IStreamPipesDataSink;
 import org.apache.streampipes.extensions.api.pe.config.IDataSinkConfiguration;
 import org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext;
diff --git a/streampipes-extensions/streampipes-connectors-rocketmq/src/main/java/org/apache/streampipes/extensions/connectors/rocketmq/sink/RocketMQPublisherSink.java b/streampipes-extensions/streampipes-connectors-rocketmq/src/main/java/org/apache/streampipes/extensions/connectors/rocketmq/sink/RocketMQPublisherSink.java
index 58aabf4e76..4e1df94272 100644
--- a/streampipes-extensions/streampipes-connectors-rocketmq/src/main/java/org/apache/streampipes/extensions/connectors/rocketmq/sink/RocketMQPublisherSink.java
+++ b/streampipes-extensions/streampipes-connectors-rocketmq/src/main/java/org/apache/streampipes/extensions/connectors/rocketmq/sink/RocketMQPublisherSink.java
@@ -20,7 +20,7 @@ package org.apache.streampipes.extensions.connectors.rocketmq.sink;
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
 import org.apache.streampipes.dataformat.SpDataFormatDefinition;
-import org.apache.streampipes.dataformat.json.JsonDataFormatDefinition;
+import org.apache.streampipes.dataformat.JsonDataFormatDefinition;
 import org.apache.streampipes.extensions.api.pe.IStreamPipesDataSink;
 import org.apache.streampipes.extensions.api.pe.config.IDataSinkConfiguration;
 import org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext;
diff --git a/streampipes-extensions/streampipes-connectors-ros/src/main/java/org/apache/streampipes/connectors/ros/sink/RosBridgeSink.java b/streampipes-extensions/streampipes-connectors-ros/src/main/java/org/apache/streampipes/connectors/ros/sink/RosBridgeSink.java
index cac8ef5f6e..4e5e9c6393 100644
--- a/streampipes-extensions/streampipes-connectors-ros/src/main/java/org/apache/streampipes/connectors/ros/sink/RosBridgeSink.java
+++ b/streampipes-extensions/streampipes-connectors-ros/src/main/java/org/apache/streampipes/connectors/ros/sink/RosBridgeSink.java
@@ -19,7 +19,7 @@
 package org.apache.streampipes.connectors.ros.sink;
 
 import org.apache.streampipes.dataformat.SpDataFormatDefinition;
-import org.apache.streampipes.dataformat.json.JsonDataFormatDefinition;
+import org.apache.streampipes.dataformat.JsonDataFormatDefinition;
 import org.apache.streampipes.extensions.api.pe.IStreamPipesDataSink;
 import org.apache.streampipes.extensions.api.pe.config.IDataSinkConfiguration;
 import org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext;
diff --git a/streampipes-extensions/streampipes-connectors-tubemq/src/main/java/org/apache/streampipes/extensions/connectors/tubemq/sink/TubeMQPublisherSink.java b/streampipes-extensions/streampipes-connectors-tubemq/src/main/java/org/apache/streampipes/extensions/connectors/tubemq/sink/TubeMQPublisherSink.java
index b9a2dbf72d..1037c26ec2 100644
--- a/streampipes-extensions/streampipes-connectors-tubemq/src/main/java/org/apache/streampipes/extensions/connectors/tubemq/sink/TubeMQPublisherSink.java
+++ b/streampipes-extensions/streampipes-connectors-tubemq/src/main/java/org/apache/streampipes/extensions/connectors/tubemq/sink/TubeMQPublisherSink.java
@@ -20,7 +20,7 @@ package org.apache.streampipes.extensions.connectors.tubemq.sink;
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
 import org.apache.streampipes.dataformat.SpDataFormatDefinition;
-import org.apache.streampipes.dataformat.json.JsonDataFormatDefinition;
+import org.apache.streampipes.dataformat.JsonDataFormatDefinition;
 import org.apache.streampipes.extensions.api.pe.IStreamPipesDataSink;
 import org.apache.streampipes.extensions.api.pe.config.IDataSinkConfiguration;
 import org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext;
diff --git a/streampipes-extensions/streampipes-extensions-all-iiot/src/main/java/org/apache/streampipes/extensions/all/iiot/AllExtensionsIIoTInit.java b/streampipes-extensions/streampipes-extensions-all-iiot/src/main/java/org/apache/streampipes/extensions/all/iiot/AllExtensionsIIoTInit.java
index 66187857a0..f0b49c1831 100644
--- a/streampipes-extensions/streampipes-extensions-all-iiot/src/main/java/org/apache/streampipes/extensions/all/iiot/AllExtensionsIIoTInit.java
+++ b/streampipes-extensions/streampipes-extensions-all-iiot/src/main/java/org/apache/streampipes/extensions/all/iiot/AllExtensionsIIoTInit.java
@@ -20,10 +20,6 @@ package org.apache.streampipes.extensions.all.iiot;
 
 import org.apache.streampipes.connect.iiot.IIoTAdaptersExtensionModuleExport;
 import org.apache.streampipes.connectors.ros.RosConnectorsModuleExport;
-import org.apache.streampipes.dataformat.cbor.CborDataFormatFactory;
-import org.apache.streampipes.dataformat.fst.FstDataFormatFactory;
-import org.apache.streampipes.dataformat.json.JsonDataFormatFactory;
-import org.apache.streampipes.dataformat.smile.SmileDataFormatFactory;
 import org.apache.streampipes.extensions.connectors.influx.InfluxConnectorsModuleExport;
 import org.apache.streampipes.extensions.connectors.kafka.KafkaConnectorsModuleExport;
 import org.apache.streampipes.extensions.connectors.mqtt.MqttConnectorsModuleExport;
@@ -89,11 +85,6 @@ public class AllExtensionsIIoTInit extends StreamPipesExtensionsServiceBase {
             new NotificationsExtensionModuleExport()
         )
         .registerRuntimeProvider(new StandaloneStreamPipesRuntimeProvider())
-        .registerMessagingFormats(
-            new JsonDataFormatFactory(),
-            new CborDataFormatFactory(),
-            new SmileDataFormatFactory(),
-            new FstDataFormatFactory())
         .registerMessagingProtocols(
             new SpKafkaProtocolFactory(),
             new SpJmsProtocolFactory(),
diff --git a/streampipes-extensions/streampipes-extensions-all-jvm/src/main/java/org/apache/streampipes/extensions/all/jvm/AllExtensionsInit.java b/streampipes-extensions/streampipes-extensions-all-jvm/src/main/java/org/apache/streampipes/extensions/all/jvm/AllExtensionsInit.java
index 19c7e551c1..8510863038 100644
--- a/streampipes-extensions/streampipes-extensions-all-jvm/src/main/java/org/apache/streampipes/extensions/all/jvm/AllExtensionsInit.java
+++ b/streampipes-extensions/streampipes-extensions-all-jvm/src/main/java/org/apache/streampipes/extensions/all/jvm/AllExtensionsInit.java
@@ -20,10 +20,6 @@ package org.apache.streampipes.extensions.all.jvm;
 import org.apache.streampipes.connect.GeneralAdaptersExtensionModuleExport;
 import org.apache.streampipes.connect.iiot.IIoTAdaptersExtensionModuleExport;
 import org.apache.streampipes.connectors.ros.RosConnectorsModuleExport;
-import org.apache.streampipes.dataformat.cbor.CborDataFormatFactory;
-import org.apache.streampipes.dataformat.fst.FstDataFormatFactory;
-import org.apache.streampipes.dataformat.json.JsonDataFormatFactory;
-import org.apache.streampipes.dataformat.smile.SmileDataFormatFactory;
 import org.apache.streampipes.extensions.connectors.influx.InfluxConnectorsModuleExport;
 import org.apache.streampipes.extensions.connectors.kafka.KafkaConnectorsModuleExport;
 import org.apache.streampipes.extensions.connectors.mqtt.MqttConnectorsModuleExport;
@@ -97,11 +93,6 @@ public class AllExtensionsInit extends StreamPipesExtensionsServiceBase {
             new NotificationsExtensionModuleExport()
         )
         .registerRuntimeProvider(new StandaloneStreamPipesRuntimeProvider())
-        .registerMessagingFormats(
-            new JsonDataFormatFactory(),
-            new CborDataFormatFactory(),
-            new SmileDataFormatFactory(),
-            new FstDataFormatFactory())
         .registerMessagingProtocols(
             new SpKafkaProtocolFactory(),
             new SpJmsProtocolFactory(),
diff --git a/streampipes-extensions/streampipes-extensions-iiot-minimal/src/main/java/org/apache/streampipes/extensions/iiot/minimal/ExtensionsIIoTMinimalInit.java b/streampipes-extensions/streampipes-extensions-iiot-minimal/src/main/java/org/apache/streampipes/extensions/iiot/minimal/ExtensionsIIoTMinimalInit.java
index 80db365175..b0e038d549 100644
--- a/streampipes-extensions/streampipes-extensions-iiot-minimal/src/main/java/org/apache/streampipes/extensions/iiot/minimal/ExtensionsIIoTMinimalInit.java
+++ b/streampipes-extensions/streampipes-extensions-iiot-minimal/src/main/java/org/apache/streampipes/extensions/iiot/minimal/ExtensionsIIoTMinimalInit.java
@@ -19,10 +19,6 @@
 package org.apache.streampipes.extensions.iiot.minimal;
 
 import org.apache.streampipes.connect.iiot.IIoTAdaptersExtensionModuleExport;
-import org.apache.streampipes.dataformat.cbor.CborDataFormatFactory;
-import org.apache.streampipes.dataformat.fst.FstDataFormatFactory;
-import org.apache.streampipes.dataformat.json.JsonDataFormatFactory;
-import org.apache.streampipes.dataformat.smile.SmileDataFormatFactory;
 import org.apache.streampipes.extensions.connectors.influx.InfluxConnectorsModuleExport;
 import org.apache.streampipes.extensions.connectors.mqtt.MqttConnectorsModuleExport;
 import org.apache.streampipes.extensions.connectors.nats.NatsConnectorsModuleExport;
@@ -74,11 +70,6 @@ public class ExtensionsIIoTMinimalInit extends StreamPipesExtensionsServiceBase
             new InternalSinksExtensionModuleExports(),
             new NotificationsExtensionModuleExport())
         .registerRuntimeProvider(new StandaloneStreamPipesRuntimeProvider())
-        .registerMessagingFormats(
-            new JsonDataFormatFactory(),
-            new CborDataFormatFactory(),
-            new SmileDataFormatFactory(),
-            new FstDataFormatFactory())
         .registerMessagingProtocols(
             new SpNatsProtocolFactory(),
             new SpMqttProtocolFactory()
diff --git a/streampipes-extensions/streampipes-pipeline-elements-experimental-flink/src/main/java/org/apache/streampipes/pe/flink/AllFlinkPipelineElementsInit.java b/streampipes-extensions/streampipes-pipeline-elements-experimental-flink/src/main/java/org/apache/streampipes/pe/flink/AllFlinkPipelineElementsInit.java
index 033e658399..3344fa55b1 100644
--- a/streampipes-extensions/streampipes-pipeline-elements-experimental-flink/src/main/java/org/apache/streampipes/pe/flink/AllFlinkPipelineElementsInit.java
+++ b/streampipes-extensions/streampipes-pipeline-elements-experimental-flink/src/main/java/org/apache/streampipes/pe/flink/AllFlinkPipelineElementsInit.java
@@ -17,10 +17,6 @@
  */
 package org.apache.streampipes.pe.flink;
 
-import org.apache.streampipes.dataformat.cbor.CborDataFormatFactory;
-import org.apache.streampipes.dataformat.fst.FstDataFormatFactory;
-import org.apache.streampipes.dataformat.json.JsonDataFormatFactory;
-import org.apache.streampipes.dataformat.smile.SmileDataFormatFactory;
 import org.apache.streampipes.extensions.management.model.SpServiceDefinition;
 import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBuilder;
 import org.apache.streampipes.messaging.jms.SpJmsProtocolFactory;
@@ -83,11 +79,6 @@ public class AllFlinkPipelineElementsInit extends StreamPipesExtensionsServiceBa
             new BoilerplateController(),
             new ElasticSearchController())
         .registerRuntimeProvider(new FlinkRuntimeProvider())
-        .registerMessagingFormats(
-            new JsonDataFormatFactory(),
-            new CborDataFormatFactory(),
-            new SmileDataFormatFactory(),
-            new FstDataFormatFactory())
         .registerMessagingProtocols(
             new SpKafkaProtocolFactory(),
             new SpJmsProtocolFactory(),
diff --git a/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/bufferrest/BufferRestPublisherSink.java b/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/bufferrest/BufferRestPublisherSink.java
index aea49f3ab7..95b794e5aa 100644
--- a/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/bufferrest/BufferRestPublisherSink.java
+++ b/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/bufferrest/BufferRestPublisherSink.java
@@ -19,8 +19,8 @@
 package org.apache.streampipes.sinks.brokers.jvm.bufferrest;
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.dataformat.JsonDataFormatDefinition;
 import org.apache.streampipes.dataformat.SpDataFormatDefinition;
-import org.apache.streampipes.dataformat.json.JsonDataFormatDefinition;
 import org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext;
 import org.apache.streampipes.model.DataSinkType;
 import org.apache.streampipes.model.graph.DataSinkDescription;
diff --git a/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/jms/JmsPublisherSink.java b/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/jms/JmsPublisherSink.java
index 60f88237bd..3d2fa15cbb 100644
--- a/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/jms/JmsPublisherSink.java
+++ b/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/jms/JmsPublisherSink.java
@@ -20,7 +20,7 @@ package org.apache.streampipes.sinks.brokers.jvm.jms;
 
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataformat.json.JsonDataFormatDefinition;
+import org.apache.streampipes.dataformat.JsonDataFormatDefinition;
 import org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext;
 import org.apache.streampipes.messaging.jms.ActiveMQPublisher;
 import org.apache.streampipes.model.DataSinkType;
diff --git a/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/rabbitmq/RabbitMqPublisherSink.java b/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/rabbitmq/RabbitMqPublisherSink.java
index fe1b264ee8..2e7de12842 100644
--- a/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/rabbitmq/RabbitMqPublisherSink.java
+++ b/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/rabbitmq/RabbitMqPublisherSink.java
@@ -19,7 +19,7 @@
 package org.apache.streampipes.sinks.brokers.jvm.rabbitmq;
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataformat.json.JsonDataFormatDefinition;
+import org.apache.streampipes.dataformat.JsonDataFormatDefinition;
 import org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext;
 import org.apache.streampipes.model.DataSinkType;
 import org.apache.streampipes.model.extensions.ExtensionAssetType;
diff --git a/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/rest/RestSink.java b/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/rest/RestSink.java
index 666bfb8be7..e09f4f7f81 100644
--- a/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/rest/RestSink.java
+++ b/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/rest/RestSink.java
@@ -20,7 +20,7 @@ package org.apache.streampipes.sinks.brokers.jvm.rest;
 
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataformat.json.JsonDataFormatDefinition;
+import org.apache.streampipes.dataformat.JsonDataFormatDefinition;
 import org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext;
 import org.apache.streampipes.model.DataSinkType;
 import org.apache.streampipes.model.extensions.ExtensionAssetType;
diff --git a/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/websocket/SocketServer.java b/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/websocket/SocketServer.java
index babd6f8a3f..65b817e95e 100644
--- a/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/websocket/SocketServer.java
+++ b/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/websocket/SocketServer.java
@@ -18,7 +18,7 @@
 
 package org.apache.streampipes.sinks.brokers.jvm.websocket;
 
-import org.apache.streampipes.dataformat.json.JsonDataFormatDefinition;
+import org.apache.streampipes.dataformat.JsonDataFormatDefinition;
 import org.apache.streampipes.model.runtime.Event;
 
 import org.java_websocket.WebSocket;
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/DefaultMessagingSettings.java b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/DefaultMessagingSettings.java
index 457ec36b14..6ba4dd0865 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/DefaultMessagingSettings.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/DefaultMessagingSettings.java
@@ -50,7 +50,6 @@ public class DefaultMessagingSettings {
         5000012,
         20,
         2,
-        Arrays.asList(SpDataFormat.JSON, SpDataFormat.CBOR, SpDataFormat.FST, SpDataFormat.SMILE),
         protocolList);
 
     defaultSettings.setJmsHost("activemq");
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/MessagingSettings.java b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/MessagingSettings.java
index edad5c1a50..874c200750 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/MessagingSettings.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/MessagingSettings.java
@@ -30,7 +30,6 @@ public class MessagingSettings {
   private Integer lingerMs;
   private Integer acks;
 
-  private List<SpDataFormat> prioritizedFormats;
   private List<SpProtocol> prioritizedProtocols;
 
   private String jmsHost;
@@ -56,13 +55,11 @@ public class MessagingSettings {
                            Integer messageMaxBytes,
                            Integer lingerMs,
                            Integer acks,
-                           List<SpDataFormat> prioritizedFormats,
                            List<SpProtocol> prioritizedProtocols) {
     this.batchSize = batchSize;
     this.messageMaxBytes = messageMaxBytes;
     this.lingerMs = lingerMs;
     this.acks = acks;
-    this.prioritizedFormats = prioritizedFormats;
     this.prioritizedProtocols = prioritizedProtocols;
     this.supportedProtocols = new ArrayList<>();
   }
@@ -103,14 +100,6 @@ public class MessagingSettings {
     this.acks = acks;
   }
 
-  public List<SpDataFormat> getPrioritizedFormats() {
-    return prioritizedFormats;
-  }
-
-  public void setPrioritizedFormats(List<SpDataFormat> prioritizedFormats) {
-    this.prioritizedFormats = prioritizedFormats;
-  }
-
   public List<SpProtocol> getPrioritizedProtocols() {
     return prioritizedProtocols;
   }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/SpDataFormat.java b/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/SpDataFormat.java
deleted file mode 100644
index a2c5b6bf38..0000000000
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/configuration/SpDataFormat.java
+++ /dev/null
@@ -1,45 +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.model.configuration;
-
-
-import org.apache.streampipes.vocabulary.MessageFormat;
-
-public enum SpDataFormat {
-
-  CBOR("Cbor", MessageFormat.CBOR),
-  JSON("JSON", MessageFormat.JSON),
-  FST("Fast-Serializer", MessageFormat.FST),
-  SMILE("Smile", MessageFormat.SMILE);
-
-  private String name;
-  private String messageFormat;
-
-  SpDataFormat(String name, String messageFormat) {
-    this.name = name;
-    this.messageFormat = messageFormat;
-  }
-
-  public String getName() {
-    return name;
-  }
-
-  public String getMessageFormat() {
-    return messageFormat;
-  }
-}
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/grounding/EventGrounding.java b/streampipes-model/src/main/java/org/apache/streampipes/model/grounding/EventGrounding.java
index ccadf9652c..f22a632b11 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/grounding/EventGrounding.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/grounding/EventGrounding.java
@@ -23,7 +23,6 @@ import org.apache.streampipes.model.util.Cloner;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 
@@ -33,24 +32,18 @@ public class EventGrounding {
 
   private List<TransportProtocol> transportProtocols;
 
-  private List<TransportFormat> transportFormats;
-
   public EventGrounding() {
     super();
-    this.transportFormats = new ArrayList<>();
     this.transportProtocols = new ArrayList<>();
   }
 
-  public EventGrounding(TransportProtocol transportProtocol, TransportFormat transportFormat) {
+  public EventGrounding(TransportProtocol transportProtocol) {
     this();
-    this.transportFormats = new ArrayList<>();
-    this.transportFormats.add(transportFormat);
-    this.transportProtocols = Arrays.asList(transportProtocol);
+    this.transportProtocols = Collections.singletonList(transportProtocol);
   }
 
   public EventGrounding(EventGrounding other) {
     this.transportProtocols = new Cloner().protocols(other.getTransportProtocols());
-    this.transportFormats = new Cloner().transportFormats(other.getTransportFormats());
   }
 
   public List<TransportProtocol> getTransportProtocols() {
@@ -63,7 +56,7 @@ public class EventGrounding {
 
   @JsonIgnore
   public TransportProtocol getTransportProtocol() {
-    if (transportProtocols.size() == 0) {
+    if (transportProtocols.isEmpty()) {
       return null;
     } else {
       return transportProtocols.get(0);
@@ -73,13 +66,4 @@ public class EventGrounding {
   public void setTransportProtocol(TransportProtocol transportProtocol) {
     this.transportProtocols = Collections.singletonList(transportProtocol);
   }
-
-  public List<TransportFormat> getTransportFormats() {
-    return transportFormats;
-  }
-
-  public void setTransportFormats(List<TransportFormat> transportFormats) {
-    this.transportFormats = transportFormats;
-  }
-
 }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/util/Cloner.java b/streampipes-model/src/main/java/org/apache/streampipes/model/util/Cloner.java
index ebdf536ce2..853415270c 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/util/Cloner.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/util/Cloner.java
@@ -30,7 +30,6 @@ import org.apache.streampipes.model.grounding.NatsTransportProtocol;
 import org.apache.streampipes.model.grounding.PulsarTransportProtocol;
 import org.apache.streampipes.model.grounding.SimpleTopicDefinition;
 import org.apache.streampipes.model.grounding.TopicDefinition;
-import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.model.grounding.TransportProtocol;
 import org.apache.streampipes.model.grounding.WildcardTopicDefinition;
 import org.apache.streampipes.model.grounding.WildcardTopicMapping;
@@ -222,11 +221,6 @@ public class Cloner {
     }
   }
 
-  public List<TransportFormat> transportFormats(
-      List<TransportFormat> transportFormats) {
-    return transportFormats.stream().map(t -> new TransportFormat(t)).collect(Collectors.toList());
-  }
-
   public List<EventProperty> properties(List<EventProperty> eventProperties) {
     return eventProperties.stream().map(o -> new Cloner().property(o)).collect(Collectors.toList());
   }
diff --git a/streampipes-pipeline-management/pom.xml b/streampipes-pipeline-management/pom.xml
index ef233e90f7..24d59b59e3 100644
--- a/streampipes-pipeline-management/pom.xml
+++ b/streampipes-pipeline-management/pom.xml
@@ -34,22 +34,7 @@
         </dependency>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-dataformat-cbor</artifactId>
-            <version>0.97.0-SNAPSHOT</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-dataformat-fst</artifactId>
-            <version>0.97.0-SNAPSHOT</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-dataformat-json</artifactId>
-            <version>0.97.0-SNAPSHOT</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-dataformat-smile</artifactId>
+            <artifactId>streampipes-dataformat</artifactId>
             <version>0.97.0-SNAPSHOT</version>
         </dependency>
         <dependency>
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/FormatSelector.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/FormatSelector.java
deleted file mode 100644
index 7b38301179..0000000000
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/FormatSelector.java
+++ /dev/null
@@ -1,83 +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.manager.matching;
-
-import org.apache.streampipes.model.SpDataStream;
-import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
-import org.apache.streampipes.model.base.NamedStreamPipesEntity;
-import org.apache.streampipes.model.configuration.SpDataFormat;
-import org.apache.streampipes.model.grounding.TransportFormat;
-import org.apache.streampipes.storage.management.StorageDispatcher;
-import org.apache.streampipes.vocabulary.MessageFormat;
-
-import java.net.URI;
-import java.util.List;
-import java.util.Set;
-import java.util.stream.Collectors;
-
-public class FormatSelector extends GroundingSelector {
-
-  public FormatSelector(NamedStreamPipesEntity source, Set<InvocableStreamPipesEntity> targets) {
-    super(source, targets);
-  }
-
-  public TransportFormat getTransportFormat() {
-
-    if (source instanceof SpDataStream) {
-      return ((SpDataStream) source)
-          .getEventGrounding()
-          .getTransportFormats()
-          .get(0);
-    } else {
-      List<SpDataFormat> prioritizedFormats =
-          StorageDispatcher
-              .INSTANCE
-              .getNoSqlStore()
-              .getSpCoreConfigurationStorage()
-              .get()
-              .getMessagingSettings()
-              .getPrioritizedFormats();
-
-      List<SpDataFormat> supportedFormats = prioritizedFormats
-          .stream()
-          .filter(pf -> supportsFormat(pf.getMessageFormat())).toList();
-
-      if (supportedFormats.size() > 0) {
-        return new TransportFormat(supportedFormats.get(0).getMessageFormat());
-      } else {
-        return new TransportFormat(MessageFormat.JSON);
-      }
-    }
-  }
-
-  public boolean supportsFormat(String format) {
-    List<InvocableStreamPipesEntity> elements = buildInvocables();
-    return elements
-        .stream()
-        .allMatch(e -> e
-            .getSupportedGrounding()
-            .getTransportFormats()
-            .stream()
-            .anyMatch(s -> rdfTypesAsString(s.getRdfType()).contains(format)));
-  }
-
-  private List<String> rdfTypesAsString(List<URI> uri) {
-    return uri.stream().map(URI::toString).collect(Collectors.toList());
-  }
-}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/GroundingBuilder.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/GroundingBuilder.java
index ee683afd93..4e711c11af 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/GroundingBuilder.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/GroundingBuilder.java
@@ -21,7 +21,6 @@ package org.apache.streampipes.manager.matching;
 import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
 import org.apache.streampipes.model.base.NamedStreamPipesEntity;
 import org.apache.streampipes.model.grounding.EventGrounding;
-import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.model.grounding.TransportProtocol;
 
 import java.util.Collections;
@@ -39,15 +38,10 @@ public class GroundingBuilder {
 
   public EventGrounding getEventGrounding() {
     EventGrounding grounding = new EventGrounding();
-    grounding.setTransportFormats(Collections.singletonList(getFormat()));
     grounding.setTransportProtocols(Collections.singletonList(getProtocol()));
     return grounding;
   }
 
-  private TransportFormat getFormat() {
-    return new FormatSelector(source, targets).getTransportFormat();
-  }
-
   private TransportProtocol getProtocol() {
     return new ProtocolSelector(source, targets).getPreferredProtocol();
   }
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/FormatMatch.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/FormatMatch.java
deleted file mode 100644
index e864cc6ea3..0000000000
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/FormatMatch.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.manager.matching.v2;
-
-import org.apache.streampipes.manager.matching.v2.utils.MatchingUtils;
-import org.apache.streampipes.model.client.matching.MatchingResultMessage;
-import org.apache.streampipes.model.client.matching.MatchingResultType;
-import org.apache.streampipes.model.grounding.TransportFormat;
-
-import java.util.List;
-
-public class FormatMatch extends AbstractMatcher<TransportFormat, TransportFormat> {
-
-  public FormatMatch() {
-    super(MatchingResultType.FORMAT_MATCH);
-  }
-
-  @Override
-  public boolean match(TransportFormat offer, TransportFormat requirement, List<MatchingResultMessage> errorLog) {
-    return MatchingUtils.nullCheck(offer, requirement)
-        || requirement.getRdfType().containsAll(offer.getRdfType());
-
-  }
-}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/GroundingMatch.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/GroundingMatch.java
index 94bd5d3255..27d403a5e2 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/GroundingMatch.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/v2/GroundingMatch.java
@@ -22,7 +22,6 @@ import org.apache.streampipes.manager.matching.v2.utils.MatchingUtils;
 import org.apache.streampipes.model.client.matching.MatchingResultMessage;
 import org.apache.streampipes.model.client.matching.MatchingResultType;
 import org.apache.streampipes.model.grounding.EventGrounding;
-import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.model.grounding.TransportProtocol;
 
 import java.util.List;
@@ -36,8 +35,7 @@ public class GroundingMatch extends AbstractMatcher<EventGrounding, EventGroundi
   @Override
   public boolean match(EventGrounding offer, EventGrounding requirement, List<MatchingResultMessage> errorLog) {
     return MatchingUtils.nullCheckRightNullDisallowed(offer, requirement)
-        || (matchProtocols(offer.getTransportProtocols(), requirement.getTransportProtocols(), errorLog)
-        && matchFormats(offer.getTransportFormats(), requirement.getTransportFormats(), errorLog));
+        || (matchProtocols(offer.getTransportProtocols(), requirement.getTransportProtocols(), errorLog));
   }
 
   private boolean matchProtocols(List<TransportProtocol> offer, List<TransportProtocol> requirement,
@@ -52,18 +50,4 @@ public class GroundingMatch extends AbstractMatcher<EventGrounding, EventGroundi
     }
     return match;
   }
-
-  private boolean matchFormats(List<TransportFormat> offer, List<TransportFormat> requirement,
-                               List<MatchingResultMessage> errorLog) {
-    boolean match = MatchingUtils.nullCheckBothNullDisallowed(offer, requirement)
-        && requirement
-            .stream()
-            .anyMatch(req -> offer.stream().anyMatch(of -> new FormatMatch().match(of, req, errorLog)));
-
-    if (!match) {
-      buildErrorMessage(errorLog, MatchingResultType.FORMAT_MATCH, "Could not find matching format");
-    }
-    return match;
-  }
-
 }
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/runtime/DataStreamRuntimeInfoProvider.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/runtime/DataStreamRuntimeInfoProvider.java
index 7221339d13..3fec8e42ae 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/runtime/DataStreamRuntimeInfoProvider.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/runtime/DataStreamRuntimeInfoProvider.java
@@ -20,11 +20,11 @@ package org.apache.streampipes.manager.runtime;
 import org.apache.streampipes.commons.environment.Environment;
 import org.apache.streampipes.commons.environment.Environments;
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.dataformat.SpDataFormatManager;
 import org.apache.streampipes.messaging.EventConsumer;
 import org.apache.streampipes.messaging.SpProtocolManager;
 import org.apache.streampipes.model.SpDataStream;
 import org.apache.streampipes.model.grounding.KafkaTransportProtocol;
-import org.apache.streampipes.model.grounding.TransportFormat;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -60,7 +60,7 @@ public class DataStreamRuntimeInfoProvider {
         }
       }
 
-      var converter = new SpDataFormatConverterGenerator(getTransportFormat(dataStream)).makeConverter();
+      var converter = new SpDataFormatConverter(SpDataFormatManager.getFormatDefinition());
       var protocolDefinitionOpt = SpProtocolManager
           .INSTANCE
           .findDefinition(dataStream.getEventGrounding().getTransportProtocol());
@@ -80,10 +80,6 @@ public class DataStreamRuntimeInfoProvider {
     });
   }
 
-  private TransportFormat getTransportFormat(SpDataStream spDataStream) {
-    return spDataStream.getEventGrounding().getTransportFormats().get(0);
-  }
-
   public Map<String, Map<String, Object>> getLatestEvents() {
     return latestEvents;
   }
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/runtime/SpDataFormatConverterGenerator.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/runtime/SpDataFormatConverterGenerator.java
deleted file mode 100644
index 2d4e57ab9d..0000000000
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/runtime/SpDataFormatConverterGenerator.java
+++ /dev/null
@@ -1,74 +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.manager.runtime;
-
-import org.apache.streampipes.dataformat.cbor.CborDataFormatDefinition;
-import org.apache.streampipes.dataformat.fst.FstDataFormatDefinition;
-import org.apache.streampipes.dataformat.json.JsonDataFormatDefinition;
-import org.apache.streampipes.dataformat.smile.SmileDataFormatDefinition;
-import org.apache.streampipes.model.grounding.TransportFormat;
-import org.apache.streampipes.vocabulary.MessageFormat;
-
-public class SpDataFormatConverterGenerator {
-
-  private TransportFormat transportFormat;
-
-  public SpDataFormatConverterGenerator(TransportFormat transportFormat) {
-    this.transportFormat = transportFormat;
-  }
-
-  public SpDataFormatConverter makeConverter() {
-    if (isJsonFormat(transportFormat)) {
-      return new SpDataFormatConverter(new JsonDataFormatDefinition());
-    } else if (isCborFormat(transportFormat)) {
-      return new SpDataFormatConverter(new CborDataFormatDefinition());
-    } else if (isFstFormat(transportFormat)) {
-      return new SpDataFormatConverter(new FstDataFormatDefinition());
-    } else if (isSmileFormat(transportFormat)) {
-      return new SpDataFormatConverter(new SmileDataFormatDefinition());
-    } else {
-      throw new IllegalArgumentException("Wrong transport format: " + makeError(transportFormat));
-    }
-  }
-
-  private boolean isSmileFormat(TransportFormat transportFormat) {
-    return isFormat(MessageFormat.SMILE, transportFormat);
-  }
-
-  private boolean isFstFormat(TransportFormat transportFormat) {
-    return isFormat(MessageFormat.FST, transportFormat);
-  }
-
-  private boolean isCborFormat(TransportFormat transportFormat) {
-    return isFormat(MessageFormat.CBOR, transportFormat);
-  }
-
-  private Boolean isJsonFormat(TransportFormat transportFormat) {
-    return isFormat(MessageFormat.JSON, transportFormat);
-  }
-
-  private boolean isFormat(String format, TransportFormat transportFormat) {
-    return transportFormat.getRdfType().stream().anyMatch(tf -> tf.toString().equals(format));
-  }
-
-  private String makeError(TransportFormat transportFormat) {
-    StringBuilder builder = new StringBuilder();
-    transportFormat.getRdfType().forEach(type -> builder.append(type.toString()).append(", "));
-    return builder.toString();
-  }
-}
diff --git a/streampipes-rest-extensions/src/main/java/org/apache/streampipes/rest/extensions/AbstractPipelineElementResource.java b/streampipes-rest-extensions/src/main/java/org/apache/streampipes/rest/extensions/AbstractPipelineElementResource.java
index e2bd86a454..b16f23b023 100644
--- a/streampipes-rest-extensions/src/main/java/org/apache/streampipes/rest/extensions/AbstractPipelineElementResource.java
+++ b/streampipes-rest-extensions/src/main/java/org/apache/streampipes/rest/extensions/AbstractPipelineElementResource.java
@@ -26,7 +26,6 @@ import org.apache.streampipes.extensions.management.locales.LabelGenerator;
 import org.apache.streampipes.model.base.ConsumableStreamPipesEntity;
 import org.apache.streampipes.model.base.NamedStreamPipesEntity;
 import org.apache.streampipes.model.grounding.EventGrounding;
-import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.model.grounding.TransportProtocol;
 
 import com.google.common.base.Charsets;
@@ -125,13 +124,11 @@ public abstract class AbstractPipelineElementResource<
       if (desc instanceof ConsumableStreamPipesEntity) {
         Collection<TransportProtocol> supportedProtocols =
             DeclarersSingleton.getInstance().getSupportedProtocols();
-        Collection<TransportFormat> supportedFormats =
-            DeclarersSingleton.getInstance().getSupportedFormats();
 
-        if (supportedProtocols.size() > 0 && supportedFormats.size() > 0) {
+        if (!supportedProtocols.isEmpty()) {
           // Overwrite existing grounding from default provided by declarers singleton
           ((ConsumableStreamPipesEntity) desc)
-              .setSupportedGrounding(makeGrounding(supportedProtocols, supportedFormats));
+              .setSupportedGrounding(makeGrounding(supportedProtocols));
         }
       }
     }
@@ -139,11 +136,9 @@ public abstract class AbstractPipelineElementResource<
     return desc;
   }
 
-  private EventGrounding makeGrounding(Collection<TransportProtocol> supportedProtocols,
-                                       Collection<TransportFormat> supportedFormats) {
+  private EventGrounding makeGrounding(Collection<TransportProtocol> supportedProtocols) {
     EventGrounding grounding = new EventGrounding();
     grounding.setTransportProtocols(new ArrayList<>(supportedProtocols));
-    grounding.setTransportFormats(new ArrayList<>(supportedFormats));
 
     return grounding;
   }
diff --git a/streampipes-sdk-bundle/pom.xml b/streampipes-sdk-bundle/pom.xml
index a3c70711cb..4a767c9b33 100644
--- a/streampipes-sdk-bundle/pom.xml
+++ b/streampipes-sdk-bundle/pom.xml
@@ -39,22 +39,7 @@
         </dependency>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-dataformat-json</artifactId>
-            <version>0.97.0-SNAPSHOT</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-dataformat-cbor</artifactId>
-            <version>0.97.0-SNAPSHOT</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-dataformat-smile</artifactId>
-            <version>0.97.0-SNAPSHOT</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-dataformat-fst</artifactId>
+            <artifactId>streampipes-dataformat</artifactId>
             <version>0.97.0-SNAPSHOT</version>
         </dependency>
         <dependency>
diff --git a/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/builder/AbstractProcessingElementBuilder.java b/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/builder/AbstractProcessingElementBuilder.java
index d99618eec1..b6d3a94492 100644
--- a/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/builder/AbstractProcessingElementBuilder.java
+++ b/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/builder/AbstractProcessingElementBuilder.java
@@ -145,11 +145,13 @@ public abstract class AbstractProcessingElementBuilder<K extends
    * Assigns supported transport formats to the pipeline elements that can be handled at runtime (e.g.,
    * JSON or XMl).
    *
+   * @deprecated format assignment is no longer necessary
    * @param format An arbitrary number of supported {@link org.apache.streampipes.model.grounding.TransportFormat}s. Use
    *               {@link org.apache.streampipes.sdk.helpers.SupportedFormats} to assign formats from some pre-defined
    *               ones or create your own by following the developer guide.
    * @return this
    */
+  @Deprecated(forRemoval = true, since = "0.97.0")
   public K supportedFormats(TransportFormat... format) {
     return supportedFormats(Arrays.asList(format));
   }
@@ -158,13 +160,14 @@ public abstract class AbstractProcessingElementBuilder<K extends
    * Assigns supported transport formats to the pipeline elements that can be handled at runtime (e.g.,
    * JSON or XMl).
    *
+   * @deprecated format assignment is no longer necessary
    * @param formats A list of supported {@link org.apache.streampipes.model.grounding.TransportFormat}s. Use
    *                {@link org.apache.streampipes.sdk.helpers.SupportedFormats} to assign formats from some pre-defined
    *                ones or create your own by following the developer guide.
    * @return this
    */
+  @Deprecated(forRemoval = true, since = "0.97.0")
   public K supportedFormats(List<TransportFormat> formats) {
-    this.supportedGrounding.setTransportFormats(formats);
     return me();
   }
 
diff --git a/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/builder/DataStreamBuilder.java b/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/builder/DataStreamBuilder.java
index e1f606c497..613a6e8eb4 100644
--- a/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/builder/DataStreamBuilder.java
+++ b/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/builder/DataStreamBuilder.java
@@ -20,13 +20,11 @@ package org.apache.streampipes.sdk.builder;
 
 import org.apache.streampipes.model.SpDataStream;
 import org.apache.streampipes.model.grounding.EventGrounding;
-import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.model.grounding.TransportProtocol;
 import org.apache.streampipes.model.schema.EventProperty;
 import org.apache.streampipes.model.schema.EventSchema;
 
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 
 public class DataStreamBuilder extends AbstractPipelineElementBuilder<DataStreamBuilder, SpDataStream> {
@@ -108,19 +106,6 @@ public class DataStreamBuilder extends AbstractPipelineElementBuilder<DataStream
     return this;
   }
 
-  /**
-   * Assigns a new {@link org.apache.streampipes.model.grounding.TransportFormat} to the stream definition.
-   *
-   * @param format The transport format of the stream at runtime (e.g., JSON or Thrift).
-   *               Use {@link org.apache.streampipes.sdk.helpers.Formats} to use some pre-defined formats
-   *               (or create a new format as described in the developer guide).
-   * @return this
-   */
-  public DataStreamBuilder format(TransportFormat format) {
-    this.eventGrounding.setTransportFormats(Collections.singletonList(format));
-    return this;
-  }
-
   @Override
   protected DataStreamBuilder me() {
     return this;
diff --git a/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/helpers/Formats.java b/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/helpers/Formats.java
deleted file mode 100644
index 88107cb2d3..0000000000
--- a/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/helpers/Formats.java
+++ /dev/null
@@ -1,70 +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.sdk.helpers;
-
-import org.apache.streampipes.model.grounding.TransportFormat;
-import org.apache.streampipes.vocabulary.MessageFormat;
-
-public class Formats {
-
-  /**
-   * Defines the transport format JSON used by a data stream at runtime.
-   *
-   * @return The {@link org.apache.streampipes.model.grounding.TransportFormat} of type JSON.
-   */
-  public static TransportFormat jsonFormat() {
-    return new TransportFormat(MessageFormat.JSON);
-  }
-
-  /**
-   * Defines the transport format CBOR used by a data stream at runtime.
-   *
-   * @return The {@link org.apache.streampipes.model.grounding.TransportFormat} of type CBOR.
-   */
-  public static TransportFormat cborFormat() {
-    return new TransportFormat(MessageFormat.CBOR);
-  }
-
-  /**
-   * Defines the transport format Fast-Serializer used by a data stream at runtime.
-   *
-   * @return The {@link org.apache.streampipes.model.grounding.TransportFormat} of type FST.
-   */
-  public static TransportFormat fstFormat() {
-    return new TransportFormat(MessageFormat.FST);
-  }
-
-  /**
-   * Defines the transport format SMILE used by a data stream at runtime.
-   *
-   * @return The {@link org.apache.streampipes.model.grounding.TransportFormat} of type SMILE.
-   */
-  public static TransportFormat smileFormat() {
-    return new TransportFormat(MessageFormat.SMILE);
-  }
-
-  /**
-   * Defines the transport format Apache Thrift used by a data stream at runtime.
-   *
-   * @return The {@link org.apache.streampipes.model.grounding.TransportFormat} of type Thrift.
-   */
-  public static TransportFormat thriftFormat() {
-    return new TransportFormat(MessageFormat.THRIFT);
-  }
-}
diff --git a/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/helpers/SupportedFormats.java b/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/helpers/SupportedFormats.java
index 0f904c4cdb..e70839e182 100644
--- a/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/helpers/SupportedFormats.java
+++ b/streampipes-sdk/src/main/java/org/apache/streampipes/sdk/helpers/SupportedFormats.java
@@ -21,6 +21,10 @@ package org.apache.streampipes.sdk.helpers;
 import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.vocabulary.MessageFormat;
 
+/**
+ * @deprecated formats are no longer necessary
+ */
+@Deprecated(forRemoval = true, since = "0.97.0")
 public class SupportedFormats {
 
   /**
diff --git a/streampipes-wrapper-distributed/src/main/java/org/apache/streampipes/wrapper/distributed/runtime/DistributedRuntime.java b/streampipes-wrapper-distributed/src/main/java/org/apache/streampipes/wrapper/distributed/runtime/DistributedRuntime.java
index 9e9b0578ce..1dcaeff0b7 100644
--- a/streampipes-wrapper-distributed/src/main/java/org/apache/streampipes/wrapper/distributed/runtime/DistributedRuntime.java
+++ b/streampipes-wrapper-distributed/src/main/java/org/apache/streampipes/wrapper/distributed/runtime/DistributedRuntime.java
@@ -33,7 +33,6 @@ import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
 import org.apache.streampipes.model.grounding.JmsTransportProtocol;
 import org.apache.streampipes.model.grounding.KafkaTransportProtocol;
 import org.apache.streampipes.model.grounding.MqttTransportProtocol;
-import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.model.grounding.TransportProtocol;
 import org.apache.streampipes.wrapper.runtime.PipelineElementRuntime;
 
@@ -61,8 +60,8 @@ public abstract class DistributedRuntime<
     return new ProducerConfigFactory(protocol).makeDefaultProperties();
   }
 
-  protected SpDataFormatDefinition getDataFormatDefinition(TransportFormat transportFormat) {
-    return SpDataFormatManager.INSTANCE.findDefinition(transportFormat).get();
+  protected SpDataFormatDefinition getDataFormatDefinition() {
+    return SpDataFormatManager.getFormatDefinition();
   }
 
   protected String getTopic(SpDataStream stream) {
diff --git a/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkDataProcessorRuntime.java b/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkDataProcessorRuntime.java
index 27c8614dfe..71ad6bafe2 100644
--- a/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkDataProcessorRuntime.java
+++ b/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkDataProcessorRuntime.java
@@ -80,7 +80,7 @@ public class FlinkDataProcessorRuntime extends FlinkRuntime<
 
     EventGrounding outputGrounding = getOutputStream().getEventGrounding();
     SpDataFormatDefinition outputDataFormatDefinition =
-        getDataFormatDefinition(outputGrounding.getTransportFormats().get(0));
+        getDataFormatDefinition();
 
     ByteArraySerializer serializer =
         new ByteArraySerializer(outputDataFormatDefinition);
diff --git a/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkRuntime.java b/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkRuntime.java
index 5b77768f4d..022668816f 100644
--- a/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkRuntime.java
+++ b/streampipes-wrapper-flink/src/main/java/org/apache/streampipes/wrapper/flink/FlinkRuntime.java
@@ -34,7 +34,6 @@ import org.apache.streampipes.model.grounding.JmsTransportProtocol;
 import org.apache.streampipes.model.grounding.KafkaTransportProtocol;
 import org.apache.streampipes.model.grounding.MqttTransportProtocol;
 import org.apache.streampipes.model.grounding.SimpleTopicDefinition;
-import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.model.grounding.TransportProtocol;
 import org.apache.streampipes.model.runtime.Event;
 import org.apache.streampipes.wrapper.distributed.runtime.DistributedRuntime;
@@ -136,8 +135,7 @@ public abstract class FlinkRuntime<
       SpDataStream stream = runtimeParameters.getModel().getInputStreams().get(i);
       if (stream != null) {
         TransportProtocol protocol = stream.getEventGrounding().getTransportProtocol();
-        TransportFormat format = stream.getEventGrounding().getTransportFormats().get(0);
-        SpDataFormatDefinition dataFormatDefinition = getDataFormatDefinition(format);
+        SpDataFormatDefinition dataFormatDefinition = getDataFormatDefinition();
         if (protocol instanceof KafkaTransportProtocol) {
           return getKafkaConsumer((KafkaTransportProtocol) protocol, dataFormatDefinition);
         } else if (protocol instanceof JmsTransportProtocol) {
diff --git a/streampipes-wrapper-kafka-streams/pom.xml b/streampipes-wrapper-kafka-streams/pom.xml
index 579b32bf99..fe6affe34c 100644
--- a/streampipes-wrapper-kafka-streams/pom.xml
+++ b/streampipes-wrapper-kafka-streams/pom.xml
@@ -34,6 +34,11 @@
             <artifactId>streampipes-wrapper-distributed</artifactId>
             <version>0.97.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-dataformat</artifactId>
+            <version>0.97.0-SNAPSHOT</version>
+        </dependency>
 
         <!-- External dependencies -->
         <dependency>
diff --git a/streampipes-wrapper-kafka-streams/src/main/java/org/apache/streampipes/wrapper/kafka/KafkaStreamsDataProcessorRuntime.java b/streampipes-wrapper-kafka-streams/src/main/java/org/apache/streampipes/wrapper/kafka/KafkaStreamsDataProcessorRuntime.java
index a76552063d..fed80cc1f1 100644
--- a/streampipes-wrapper-kafka-streams/src/main/java/org/apache/streampipes/wrapper/kafka/KafkaStreamsDataProcessorRuntime.java
+++ b/streampipes-wrapper-kafka-streams/src/main/java/org/apache/streampipes/wrapper/kafka/KafkaStreamsDataProcessorRuntime.java
@@ -103,10 +103,7 @@ public class KafkaStreamsDataProcessorRuntime extends KafkaStreamsRuntime<
   }
 
   private SpDataFormatDefinition getDataFormatConverter() {
-    return SpDataFormatManager
-        .INSTANCE
-        .findDefinition(pipelineElementInvocation.getOutputStream().getEventGrounding().getTransportFormats().get(0))
-        .orElseThrow();
+    return SpDataFormatManager.getFormatDefinition();
   }
 
   private SpProtocolDefinition<KafkaTransportProtocol> getOutputProtocol() {
diff --git a/streampipes-wrapper-standalone/pom.xml b/streampipes-wrapper-standalone/pom.xml
index f4ac11cb32..e80cce593c 100644
--- a/streampipes-wrapper-standalone/pom.xml
+++ b/streampipes-wrapper-standalone/pom.xml
@@ -34,6 +34,11 @@
             <artifactId>streampipes-wrapper</artifactId>
             <version>0.97.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-dataformat</artifactId>
+            <version>0.97.0-SNAPSHOT</version>
+        </dependency>
     </dependencies>
     <build>
         <plugins>
@@ -43,4 +48,4 @@
             </plugin>
         </plugins>
     </build>
-</project>
\ No newline at end of file
+</project>
diff --git a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/function/StreamPipesFunction.java b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/function/StreamPipesFunction.java
index bd3b788c97..857a26312f 100644
--- a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/function/StreamPipesFunction.java
+++ b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/function/StreamPipesFunction.java
@@ -152,7 +152,6 @@ public abstract class StreamPipesFunction implements IStreamPipesFunctionDeclare
           uniqueStreamId,
           ProtocolManager.makeOutputCollector(
               value.getEventGrounding().getTransportProtocol(),
-              value.getEventGrounding().getTransportFormats().get(0),
               uniqueStreamId));
     });
 
@@ -169,7 +168,7 @@ public abstract class StreamPipesFunction implements IStreamPipesFunctionDeclare
         GroundingDebugUtils.modifyGrounding(is.getEventGrounding());
       }
       inputCollectors.put(uniqueStreamId, ProtocolManager.findInputCollector(is.getEventGrounding()
-          .getTransportProtocol(), is.getEventGrounding().getTransportFormats().get(0), false));
+          .getTransportProtocol(), false));
     }
     return inputCollectors;
   }
diff --git a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/manager/PManager.java b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/manager/PManager.java
index 1701566651..b94f63ea94 100644
--- a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/manager/PManager.java
+++ b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/manager/PManager.java
@@ -23,7 +23,6 @@ import org.apache.streampipes.dataformat.SpDataFormatDefinition;
 import org.apache.streampipes.dataformat.SpDataFormatManager;
 import org.apache.streampipes.messaging.SpProtocolDefinition;
 import org.apache.streampipes.messaging.SpProtocolManager;
-import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.model.grounding.TransportProtocol;
 
 import java.util.Optional;
@@ -34,7 +33,7 @@ public class PManager {
     return SpProtocolManager.INSTANCE.findDefinition(protocol);
   }
 
-  public static Optional<SpDataFormatDefinition> getDataFormat(TransportFormat format) throws SpRuntimeException {
-    return SpDataFormatManager.INSTANCE.findDefinition(format);
+  public static SpDataFormatDefinition getDataFormat() throws SpRuntimeException {
+    return SpDataFormatManager.getFormatDefinition();
   }
 }
diff --git a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/manager/ProtocolManager.java b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/manager/ProtocolManager.java
index 19c9b1b79d..9b5a4f3e43 100644
--- a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/manager/ProtocolManager.java
+++ b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/manager/ProtocolManager.java
@@ -19,7 +19,6 @@
 package org.apache.streampipes.wrapper.standalone.manager;
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.model.grounding.TransportProtocol;
 import org.apache.streampipes.wrapper.standalone.routing.StandaloneSpInputCollector;
 import org.apache.streampipes.wrapper.standalone.routing.StandaloneSpOutputCollector;
@@ -41,14 +40,13 @@ public class ProtocolManager {
   // in empire serializers
 
   public static <T extends TransportProtocol> StandaloneSpInputCollector findInputCollector(T protocol,
-                                                                                            TransportFormat format,
                                                                                             Boolean singletonEngine)
       throws SpRuntimeException {
 
     if (consumers.containsKey(topicName(protocol))) {
       return consumers.get(topicName(protocol));
     } else {
-      consumers.put(topicName(protocol), makeInputCollector(protocol, format, singletonEngine));
+      consumers.put(topicName(protocol), makeInputCollector(protocol, singletonEngine));
       LOG.info("Adding new consumer to consumer map (size=" + consumers.size() + "): " + topicName(protocol));
       return consumers.get(topicName(protocol));
     }
@@ -56,14 +54,13 @@ public class ProtocolManager {
   }
 
   public static <T extends TransportProtocol> StandaloneSpOutputCollector findOutputCollector(T protocol,
-                                                                                              TransportFormat format,
                                                                                               String resourceId)
       throws SpRuntimeException {
 
     if (producers.containsKey(topicName(protocol))) {
       return producers.get(topicName(protocol));
     } else {
-      producers.put(topicName(protocol), makeOutputCollector(protocol, format, resourceId));
+      producers.put(topicName(protocol), makeOutputCollector(protocol, resourceId));
       LOG.info("Adding new producer to producer map (size=" + producers.size() + "): " + topicName
           (protocol));
       return producers.get(topicName(protocol));
@@ -72,17 +69,15 @@ public class ProtocolManager {
   }
 
   private static <T extends TransportProtocol> StandaloneSpInputCollector<T> makeInputCollector(T protocol,
-                                                                                                TransportFormat format,
                                                                                                 Boolean singletonEngine)
       throws SpRuntimeException {
-    return new StandaloneSpInputCollector<>(protocol, format, singletonEngine);
+    return new StandaloneSpInputCollector<>(protocol, singletonEngine);
   }
 
   public static <T extends TransportProtocol> StandaloneSpOutputCollector<T> makeOutputCollector(T protocol,
-                                                                                                 TransportFormat format,
                                                                                                  String resourceId)
       throws SpRuntimeException {
-    return new StandaloneSpOutputCollector<>(protocol, format, resourceId);
+    return new StandaloneSpOutputCollector<>(protocol, resourceId);
   }
 
   private static String topicName(TransportProtocol protocol) {
diff --git a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/routing/StandaloneSpCollector.java b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/routing/StandaloneSpCollector.java
index a459506149..5a00389200 100644
--- a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/routing/StandaloneSpCollector.java
+++ b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/routing/StandaloneSpCollector.java
@@ -22,7 +22,6 @@ import org.apache.streampipes.commons.exceptions.SpRuntimeException;
 import org.apache.streampipes.dataformat.SpDataFormatDefinition;
 import org.apache.streampipes.extensions.api.pe.routing.PipelineElementCollector;
 import org.apache.streampipes.messaging.SpProtocolDefinition;
-import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.model.grounding.TransportProtocol;
 import org.apache.streampipes.wrapper.standalone.manager.PManager;
 
@@ -37,18 +36,15 @@ public abstract class StandaloneSpCollector<T extends TransportProtocol, W> impl
   protected T transportProtocol;
   protected SpProtocolDefinition<T> protocolDefinition;
 
-  protected TransportFormat transportFormat;
   protected SpDataFormatDefinition dataFormatDefinition;
   protected String topic;
 
 
-  public StandaloneSpCollector(T protocol, TransportFormat format) throws SpRuntimeException {
+  public StandaloneSpCollector(T protocol) throws SpRuntimeException {
     this.transportProtocol = protocol;
     this.protocolDefinition = PManager.getProtocolDefinition(protocol).orElseThrow(() -> new
         SpRuntimeException("Could not find protocol"));
-    this.transportFormat = format;
-    this.dataFormatDefinition = PManager.getDataFormat(format).orElseThrow(() -> new
-        SpRuntimeException("Could not find format"));
+    this.dataFormatDefinition = PManager.getDataFormat();
     this.consumers = new ConcurrentHashMap<>();
     this.topic = transportProtocol.getTopicDefinition().getActualTopicName();
   }
diff --git a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/routing/StandaloneSpInputCollector.java b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/routing/StandaloneSpInputCollector.java
index 890f16da20..f2ce71ee36 100644
--- a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/routing/StandaloneSpInputCollector.java
+++ b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/routing/StandaloneSpInputCollector.java
@@ -23,7 +23,6 @@ import org.apache.streampipes.extensions.api.pe.routing.RawDataProcessor;
 import org.apache.streampipes.extensions.api.pe.routing.SpInputCollector;
 import org.apache.streampipes.messaging.EventConsumer;
 import org.apache.streampipes.messaging.InternalEventProcessor;
-import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.model.grounding.TransportProtocol;
 import org.apache.streampipes.wrapper.standalone.manager.ProtocolManager;
 
@@ -36,9 +35,8 @@ public class StandaloneSpInputCollector<T extends TransportProtocol> extends
   private final EventConsumer consumer;
 
   public StandaloneSpInputCollector(T protocol,
-                                    TransportFormat format,
                                     Boolean singletonEngine) throws SpRuntimeException {
-    super(protocol, format);
+    super(protocol);
     this.consumer = protocolDefinition.getConsumer(protocol);
     this.singletonEngine = singletonEngine;
   }
@@ -66,7 +64,7 @@ public class StandaloneSpInputCollector<T extends TransportProtocol> extends
   @Override
   public void disconnect() throws SpRuntimeException {
     if (consumer.isConnected()) {
-      if (consumers.size() == 0) {
+      if (consumers.isEmpty()) {
         consumer.disconnect();
         ProtocolManager.removeInputCollector(transportProtocol);
       }
diff --git a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/routing/StandaloneSpOutputCollector.java b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/routing/StandaloneSpOutputCollector.java
index 671de27820..a228245069 100644
--- a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/routing/StandaloneSpOutputCollector.java
+++ b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/routing/StandaloneSpOutputCollector.java
@@ -24,7 +24,6 @@ import org.apache.streampipes.extensions.api.pe.routing.SpOutputCollector;
 import org.apache.streampipes.extensions.management.monitoring.ExtensionsLogger;
 import org.apache.streampipes.messaging.EventProducer;
 import org.apache.streampipes.messaging.InternalEventProcessor;
-import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.model.grounding.TransportProtocol;
 import org.apache.streampipes.model.runtime.Event;
 import org.apache.streampipes.model.runtime.EventConverter;
@@ -46,9 +45,8 @@ public class StandaloneSpOutputCollector<T extends TransportProtocol> extends
   private final ExtensionsLogger extensionsLogger;
 
   public StandaloneSpOutputCollector(T protocol,
-                                     TransportFormat format,
                                      String resourceId) throws SpRuntimeException {
-    super(protocol, format);
+    super(protocol);
     this.producer = protocolDefinition.getProducer(protocol);
     this.resourceId = resourceId;
     this.extensionsLogger = new ExtensionsLogger(resourceId);
diff --git a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventProcessorRuntime.java b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventProcessorRuntime.java
index 356f57b8c7..b9ce761f59 100644
--- a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventProcessorRuntime.java
+++ b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventProcessorRuntime.java
@@ -58,12 +58,6 @@ public class StandaloneEventProcessorRuntime extends StandalonePipelineElementRu
             .getOutputStream()
             .getEventGrounding()
             .getTransportProtocol(),
-        runtimeParameters
-            .getModel()
-            .getOutputStream()
-            .getEventGrounding()
-            .getTransportFormats()
-            .get(0),
         this.instanceId);
   }
 
diff --git a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandalonePipelineElementRuntime.java b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandalonePipelineElementRuntime.java
index a8e48596c0..e906161004 100644
--- a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandalonePipelineElementRuntime.java
+++ b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandalonePipelineElementRuntime.java
@@ -93,7 +93,7 @@ public abstract class StandalonePipelineElementRuntime<
     List<SpInputCollector> inputCollectors = new ArrayList<>();
     for (SpDataStream is : inputStreams) {
       inputCollectors.add(ProtocolManager.findInputCollector(is.getEventGrounding()
-              .getTransportProtocol(), is.getEventGrounding().getTransportFormats().get(0),
+              .getTransportProtocol(),
           false));
     }
     return inputCollectors;
diff --git a/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts b/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
index 700f825675..c84f23c159 100644
--- a/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
+++ b/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
@@ -20,7 +20,7 @@
 /* tslint:disable */
 /* eslint-disable */
 // @ts-nocheck
-// Generated using typescript-generator version 3.2.1263 on 2024-08-20 13:02:30.
+// Generated using typescript-generator version 3.2.1263 on 2024-08-27 18:35:51.
 
 export class NamedStreamPipesEntity implements Storable {
     '@class':
@@ -1456,7 +1456,6 @@ export class ErrorMessage extends Message {
 }
 
 export class EventGrounding {
-    transportFormats: TransportFormat[];
     transportProtocols: TransportProtocolUnion[];
 
     static fromData(
@@ -1467,9 +1466,6 @@ export class EventGrounding {
             return data;
         }
         const instance = target || new EventGrounding();
-        instance.transportFormats = __getCopyArrayFn(TransportFormat.fromData)(
-            data.transportFormats,
-        );
         instance.transportProtocols = __getCopyArrayFn(
             TransportProtocol.fromDataUnion,
         )(data.transportProtocols);
@@ -2287,7 +2283,6 @@ export class MessagingSettings {
     mqttPort: number;
     natsHost: string;
     natsPort: number;
-    prioritizedFormats: SpDataFormat[];
     prioritizedProtocols: SpProtocol[];
     pulsarUrl: string;
     supportedProtocols: string[];
@@ -2314,9 +2309,6 @@ export class MessagingSettings {
         instance.mqttPort = data.mqttPort;
         instance.natsHost = data.natsHost;
         instance.natsPort = data.natsPort;
-        instance.prioritizedFormats = __getCopyArrayFn(
-            __identity<SpDataFormat>(),
-        )(data.prioritizedFormats);
         instance.prioritizedProtocols = __getCopyArrayFn(
             __identity<SpProtocol>(),
         )(data.prioritizedProtocols);
@@ -3828,22 +3820,6 @@ export class TransformOutputStrategy extends OutputStrategy {
     }
 }
 
-export class TransportFormat {
-    rdfType: string[];
-
-    static fromData(
-        data: TransportFormat,
-        target?: TransportFormat,
-    ): TransportFormat {
-        if (!data) {
-            return data;
-        }
-        const instance = target || new TransportFormat();
-        instance.rdfType = __getCopyArrayFn(__identity<string>())(data.rdfType);
-        return instance;
-    }
-}
-
 export class TreeInputNode {
     children: TreeInputNode[];
     dataNode: boolean;
@@ -4049,8 +4025,6 @@ export type SelectionStaticPropertyUnion =
     | AnyStaticProperty
     | OneOfStaticProperty;
 
-export type SpDataFormat = 'CBOR' | 'JSON' | 'FST' | 'SMILE';
-
 export type SpLogLevel = 'INFO' | 'WARN' | 'ERROR';
 
 export type SpProtocol = 'KAFKA' | 'JMS' | 'MQTT' | 'NATS' | 'PULSAR';
diff --git a/ui/src/app/configuration/messaging-configuration/messaging-configuration.component.html b/ui/src/app/configuration/messaging-configuration/messaging-configuration.component.html
index f31c5fba3f..38c5eee17f 100644
--- a/ui/src/app/configuration/messaging-configuration/messaging-configuration.component.html
+++ b/ui/src/app/configuration/messaging-configuration/messaging-configuration.component.html
@@ -86,48 +86,6 @@
         </sp-split-section>
         <mat-divider></mat-divider>
 
-        <sp-split-section
-            title="Message Formats"
-            subtitle="Manage the priority of message formats used"
-        >
-            <div
-                fxFlex="100"
-                fxLayout="column"
-                fxLayoutAlign="start start"
-                class="page-container-padding-inner"
-            >
-                <div
-                    cdkDropList
-                    class="data-format-list"
-                    (cdkDropListDropped)="drop($event)"
-                    *ngIf="loadingCompleted"
-                >
-                    <div
-                        class="data-format-box"
-                        *ngFor="
-                            let format of messagingSettings.prioritizedFormats
-                        "
-                        cdkDrag
-                    >
-                        {{ format }}
-                    </div>
-                </div>
-                <div fxLayoutAlign="start center" class="mt-10">
-                    <button
-                        mat-raised-button
-                        color="accent"
-                        type="submit"
-                        class="md-raised md-primary submit-button"
-                        (click)="updateMessagingSettings()"
-                    >
-                        Update
-                    </button>
-                </div>
-            </div>
-        </sp-split-section>
-
-        <mat-divider></mat-divider>
-
         <sp-split-section
             title="Protocols"
             subtitle="Manage the priority of protocols used"
diff --git a/ui/src/app/configuration/messaging-configuration/messaging-configuration.component.ts b/ui/src/app/configuration/messaging-configuration/messaging-configuration.component.ts
index 292fb1008e..1ab319e3e2 100644
--- a/ui/src/app/configuration/messaging-configuration/messaging-configuration.component.ts
+++ b/ui/src/app/configuration/messaging-configuration/messaging-configuration.component.ts
@@ -61,14 +61,6 @@ export class MessagingConfigurationComponent implements OnInit {
             .subscribe(response => this.getMessagingSettings());
     }
 
-    drop(event: CdkDragDrop<string[]>) {
-        moveItemInArray(
-            this.messagingSettings.prioritizedFormats,
-            event.previousIndex,
-            event.currentIndex,
-        );
-    }
-
     dropProtocol(event: CdkDragDrop<string[]>) {
         moveItemInArray(
             this.messagingSettings.prioritizedProtocols,


Test patch:
diff --git a/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInfluxTest.java b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInfluxTest.java
index 3ddd6f7007..d8e95bb1b0 100644
--- a/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInfluxTest.java
+++ b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/TimeSeriesStorageInfluxTest.java
@@ -19,6 +19,7 @@
 package org.apache.streampipes.dataexplorer.influx;
 
 
+import org.apache.streampipes.commons.environment.Environment;
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
 import org.apache.streampipes.dataexplorer.influx.client.InfluxClientProvider;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
@@ -383,7 +384,7 @@ public class TimeSeriesStorageInfluxTest {
     );
 
     var influxClientProviderMock = Mockito.mock(InfluxClientProvider.class);
-    Mockito.when(influxClientProviderMock.getInitializedInfluxDBClient(ArgumentMatchers.any()))
+    Mockito.when(influxClientProviderMock.getSetUpInfluxDBClient((Environment) ArgumentMatchers.any()))
            .thenReturn(influxDBMock);
 
     return new TimeSeriesStorageInflux(measure, null, influxClientProviderMock);
diff --git a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/client/ClientLiveDataTesterBase.java b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/client/ClientLiveDataTesterBase.java
index 857191ca41..873a3092b5 100644
--- a/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/client/ClientLiveDataTesterBase.java
+++ b/streampipes-integration-tests/src/test/java/org/apache/streampipes/integration/client/ClientLiveDataTesterBase.java
@@ -25,9 +25,7 @@ import org.apache.streampipes.client.credentials.StreamPipesApiKeyCredentials;
 import org.apache.streampipes.integration.utils.Utils;
 import org.apache.streampipes.model.SpDataStream;
 import org.apache.streampipes.model.grounding.EventGrounding;
-import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.model.grounding.TransportProtocol;
-import org.apache.streampipes.vocabulary.MessageFormat;
 
 import org.testcontainers.shaded.com.google.common.collect.Maps;
 
@@ -89,7 +87,6 @@ public abstract class ClientLiveDataTesterBase<T extends TransportProtocol> impl
   private EventGrounding makeEventGrounding() {
     var grounding = new EventGrounding();
     grounding.setTransportProtocol(makeProtocol());
-    grounding.setTransportFormats(List.of(new TransportFormat(MessageFormat.JSON)));
 
     return grounding;
   }
diff --git a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/v2/TestFormatMatch.java b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/v2/TestFormatMatch.java
deleted file mode 100644
index e753961da0..0000000000
--- a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/v2/TestFormatMatch.java
+++ /dev/null
@@ -1,55 +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.manager.matching.v2;
-
-import org.apache.streampipes.model.client.matching.MatchingResultMessage;
-import org.apache.streampipes.model.grounding.TransportFormat;
-
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-
-import java.util.ArrayList;
-import java.util.List;
-
-public class TestFormatMatch {
-
-  @Test
-  public void testPositiveFormatMatch() {
-
-    TransportFormat offeredJson = TestUtils.jsonFormat();
-    TransportFormat requiredJson = TestUtils.jsonFormat();
-
-    List<MatchingResultMessage> errorLog = new ArrayList<>();
-
-    boolean matches = new FormatMatch().match(offeredJson, requiredJson, errorLog);
-    Assertions.assertTrue(matches);
-  }
-
-  @Test
-  public void testNegativeFormatMatch() {
-
-    TransportFormat offeredJson = TestUtils.jsonFormat();
-    TransportFormat requiredThrift = TestUtils.thriftFormat();
-
-    List<MatchingResultMessage> errorLog = new ArrayList<>();
-
-    boolean matches = new FormatMatch().match(offeredJson, requiredThrift, errorLog);
-    Assertions.assertFalse(matches);
-  }
-}
diff --git a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/v2/TestGroundingMatch.java b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/v2/TestGroundingMatch.java
index 67e55e5c1b..2bc0553141 100644
--- a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/v2/TestGroundingMatch.java
+++ b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/v2/TestGroundingMatch.java
@@ -20,7 +20,6 @@ package org.apache.streampipes.manager.matching.v2;
 
 import org.apache.streampipes.model.client.matching.MatchingResultMessage;
 import org.apache.streampipes.model.grounding.EventGrounding;
-import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.model.grounding.TransportProtocol;
 
 import org.junit.jupiter.api.Assertions;
@@ -37,11 +36,8 @@ public class TestGroundingMatch {
     TransportProtocol offerProtocol = TestUtils.kafkaProtocol();
     TransportProtocol requirementProtocol = TestUtils.kafkaProtocol();
 
-    TransportFormat offerFormat = TestUtils.jsonFormat();
-    TransportFormat requirementFormat = TestUtils.jsonFormat();
-
-    EventGrounding offeredGrounding = new EventGrounding(offerProtocol, offerFormat);
-    EventGrounding requiredGrounding = new EventGrounding(requirementProtocol, requirementFormat);
+    EventGrounding offeredGrounding = new EventGrounding(offerProtocol);
+    EventGrounding requiredGrounding = new EventGrounding(requirementProtocol);
 
     List<MatchingResultMessage> errorLog = new ArrayList<>();
 
@@ -55,29 +51,8 @@ public class TestGroundingMatch {
     TransportProtocol offerProtocol = TestUtils.kafkaProtocol();
     TransportProtocol requirementProtocol = TestUtils.jmsProtocol();
 
-    TransportFormat offerFormat = TestUtils.jsonFormat();
-    TransportFormat requirementFormat = TestUtils.jsonFormat();
-
-    EventGrounding offeredGrounding = new EventGrounding(offerProtocol, offerFormat);
-    EventGrounding requiredGrounding = new EventGrounding(requirementProtocol, requirementFormat);
-
-    List<MatchingResultMessage> errorLog = new ArrayList<>();
-
-    boolean matches = new GroundingMatch().match(offeredGrounding, requiredGrounding, errorLog);
-    Assertions.assertFalse(matches);
-  }
-
-  @Test
-  public void testNegativeGroundingMatchFormat() {
-
-    TransportProtocol offerProtocol = TestUtils.kafkaProtocol();
-    TransportProtocol requirementProtocol = TestUtils.kafkaProtocol();
-
-    TransportFormat offerFormat = TestUtils.jsonFormat();
-    TransportFormat requirementFormat = TestUtils.thriftFormat();
-
-    EventGrounding offeredGrounding = new EventGrounding(offerProtocol, offerFormat);
-    EventGrounding requiredGrounding = new EventGrounding(requirementProtocol, requirementFormat);
+    EventGrounding offeredGrounding = new EventGrounding(offerProtocol);
+    EventGrounding requiredGrounding = new EventGrounding(requirementProtocol);
 
     List<MatchingResultMessage> errorLog = new ArrayList<>();
 
@@ -91,11 +66,8 @@ public class TestGroundingMatch {
     TransportProtocol offerProtocol = TestUtils.kafkaProtocol();
     TransportProtocol requirementProtocol = TestUtils.jmsProtocol();
 
-    TransportFormat offerFormat = TestUtils.jsonFormat();
-    TransportFormat requirementFormat = TestUtils.thriftFormat();
-
-    EventGrounding offeredGrounding = new EventGrounding(offerProtocol, offerFormat);
-    EventGrounding requiredGrounding = new EventGrounding(requirementProtocol, requirementFormat);
+    EventGrounding offeredGrounding = new EventGrounding(offerProtocol);
+    EventGrounding requiredGrounding = new EventGrounding(requirementProtocol);
 
     List<MatchingResultMessage> errorLog = new ArrayList<>();
 
diff --git a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/v2/TestUtils.java b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/v2/TestUtils.java
index d1e31832e4..78d291a369 100644
--- a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/v2/TestUtils.java
+++ b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/v2/TestUtils.java
@@ -20,9 +20,7 @@ package org.apache.streampipes.manager.matching.v2;
 
 import org.apache.streampipes.model.grounding.JmsTransportProtocol;
 import org.apache.streampipes.model.grounding.KafkaTransportProtocol;
-import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.model.grounding.TransportProtocol;
-import org.apache.streampipes.vocabulary.MessageFormat;
 
 public class TestUtils {
 
@@ -33,13 +31,4 @@ public class TestUtils {
   public static TransportProtocol jmsProtocol() {
     return new JmsTransportProtocol("localhost", 61616, "abc");
   }
-
-  public static TransportFormat jsonFormat() {
-    return new TransportFormat(MessageFormat.JSON);
-  }
-
-  public static TransportFormat thriftFormat() {
-    return new TransportFormat(MessageFormat.THRIFT);
-  }
-
 }
diff --git a/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/grounding/EventGroundingGenerator.java b/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/grounding/EventGroundingGenerator.java
index 4f98071bfc..3e0199882f 100644
--- a/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/grounding/EventGroundingGenerator.java
+++ b/streampipes-test-utils/src/main/java/org/apache/streampipes/test/generator/grounding/EventGroundingGenerator.java
@@ -18,16 +18,12 @@
 package org.apache.streampipes.test.generator.grounding;
 
 import org.apache.streampipes.model.grounding.EventGrounding;
-import org.apache.streampipes.model.grounding.TransportFormat;
 import org.apache.streampipes.test.generator.grounding.protocol.ProtocolGenerator;
 
-import java.util.Arrays;
-
 public class EventGroundingGenerator {
 
   public static EventGrounding makeDummyGrounding() {
     EventGrounding grounding = new EventGrounding();
-    grounding.setTransportFormats(Arrays.asList(new TransportFormat()));
     grounding.setTransportProtocol(ProtocolGenerator.makeDummyProtocol());
 
     return grounding;