diff --git a/modules/camel-ext/pom.xml b/modules/camel-ext/pom.xml
index c2f89b1..9f47d19 100644
--- a/modules/camel-ext/pom.xml
+++ b/modules/camel-ext/pom.xml
@@ -95,9 +95,9 @@
         </dependency>
 
         <dependency>
-            <groupId>com.squareup.okhttp</groupId>
-            <artifactId>okhttp</artifactId>
-            <version>${okhttp.version}</version>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-http</artifactId>
+            <version>${camel.version}</version>
             <scope>test</scope>
         </dependency>
 
diff --git a/modules/camel-ext/src/main/java/org/apache/ignite/stream/camel/CamelStreamer.java b/modules/camel-ext/src/main/java/org/apache/ignite/stream/camel/CamelStreamer.java
index 369d2a6..51f3ba4 100644
--- a/modules/camel-ext/src/main/java/org/apache/ignite/stream/camel/CamelStreamer.java
+++ b/modules/camel-ext/src/main/java/org/apache/ignite/stream/camel/CamelStreamer.java
@@ -17,8 +17,6 @@
 
 package org.apache.ignite.stream.camel;
 
-import java.util.Map;
-
 import org.apache.camel.CamelContext;
 import org.apache.camel.Consumer;
 import org.apache.camel.Endpoint;
@@ -26,8 +24,9 @@
 import org.apache.camel.Processor;
 import org.apache.camel.ServiceStatus;
 import org.apache.camel.impl.DefaultCamelContext;
-import org.apache.camel.util.CamelContextHelper;
-import org.apache.camel.util.ServiceHelper;
+import org.apache.camel.support.CamelContextHelper;
+import org.apache.camel.support.service.BaseService;
+import org.apache.camel.support.service.ServiceHelper;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.internal.util.typedef.internal.A;
@@ -36,6 +35,8 @@
 import org.apache.ignite.stream.StreamMultipleTupleExtractor;
 import org.apache.ignite.stream.StreamSingleTupleExtractor;
 
+import java.util.Map;
+
 /**
  * This streamer consumes messages from an Apache Camel consumer endpoint and feeds them into an Ignite data streamer.
  *
@@ -84,9 +85,12 @@
         if (camelCtx == null)
             camelCtx = new DefaultCamelContext();
 
-        // If the Camel Context is starting or started, reject this call to start.
-        if (camelCtx.getStatus() == ServiceStatus.Started || camelCtx.getStatus() == ServiceStatus.Starting)
-            throw new IgniteException("Failed to start Camel streamer (CamelContext already started or starting).");
+        // If the camel context is not started then simply start it up
+        if (!camelCtx.isStarted())
+            camelCtx.start();
+
+        if (!camelCtx.isRunAllowed())
+            throw new IgniteException("Failed to start Camel streamer (CamelContext not in a runnable state).");
 
         log = getIgnite().log();
 
@@ -97,6 +101,7 @@
         catch (Exception e) {
             U.error(log, e);
 
+            camelCtx.stop();
             throw new IgniteException("Failed to start Camel streamer [errMsg=" + e.getMessage() + ']');
         }
 
@@ -107,16 +112,18 @@
         catch (Exception e) {
             U.error(log, e);
 
+            camelCtx.stop();
             throw new IgniteException("Failed to start Camel streamer [errMsg=" + e.getMessage() + ']');
         }
 
         // Start the Camel services.
         try {
-            ServiceHelper.startServices(camelCtx, endpoint, consumer);
+            ServiceHelper.startService(camelCtx, endpoint, consumer);
         }
         catch (Exception e) {
             U.error(log, e);
 
+            camelCtx.stop();
             try {
                 ServiceHelper.stopAndShutdownServices(camelCtx, endpoint, consumer);
 
diff --git a/modules/camel-ext/src/test/java/org/apache/ignite/stream/camel/IgniteCamelStreamerTest.java b/modules/camel-ext/src/test/java/org/apache/ignite/stream/camel/IgniteCamelStreamerTest.java
index 9858357..329f3ed 100644
--- a/modules/camel-ext/src/test/java/org/apache/ignite/stream/camel/IgniteCamelStreamerTest.java
+++ b/modules/camel-ext/src/test/java/org/apache/ignite/stream/camel/IgniteCamelStreamerTest.java
@@ -17,24 +17,8 @@
 
 package org.apache.ignite.stream.camel;
 
-import java.io.IOException;
-import java.net.ServerSocket;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.TreeSet;
-import java.util.UUID;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
 import com.google.common.base.Splitter;
 import com.google.common.collect.Lists;
-import com.squareup.okhttp.MediaType;
-import com.squareup.okhttp.OkHttpClient;
-import com.squareup.okhttp.Request;
-import com.squareup.okhttp.RequestBody;
-import com.squareup.okhttp.Response;
 import org.apache.camel.CamelContext;
 import org.apache.camel.Endpoint;
 import org.apache.camel.Exchange;
@@ -42,6 +26,7 @@
 import org.apache.camel.ServiceStatus;
 import org.apache.camel.component.properties.PropertiesComponent;
 import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.impl.engine.DefaultProducerTemplate;
 import org.apache.camel.support.LifecycleStrategySupport;
 import org.apache.ignite.Ignite;
 import org.apache.ignite.IgniteCache;
@@ -60,18 +45,29 @@
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
 import org.junit.Test;
 
+import java.net.ServerSocket;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
 import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT;
 
 /**
  * Test class for {@link CamelStreamer}.
  */
 public class IgniteCamelStreamerTest extends GridCommonAbstractTest {
-    /** text/plain media type. */
-    private static final MediaType TEXT_PLAIN = MediaType.parse("text/plain;charset=utf-8");
-
     /** The test data. */
     private static final Map<Integer, String> TEST_DATA = new HashMap<>();
 
+    /** The producer template used to publish data to the endpoint. */
+    private DefaultProducerTemplate producerTemplate;
+
     /** The Camel streamer currently under test. */
     private CamelStreamer<Integer, String> streamer;
 
@@ -84,9 +80,6 @@
     /** The UUID of the currently active remote listener. */
     private UUID remoteLsnr;
 
-    /** The OkHttpClient. */
-    private OkHttpClient httpClient = new OkHttpClient();
-
     // Initialize the test data.
     static {
         for (int i = 0; i < 100; i++)
@@ -114,9 +107,13 @@
             url = "http://localhost:" + port + "/ignite";
         }
 
+        DefaultCamelContext context = new DefaultCamelContext();
         // create Camel streamer
         dataStreamer = grid().dataStreamer(DEFAULT_CACHE_NAME);
         streamer = createCamelStreamer(dataStreamer);
+        streamer.setCamelContext(context);
+        producerTemplate = new DefaultProducerTemplate(context);
+        producerTemplate.start();
     }
 
     @Override public void afterTest() throws Exception {
@@ -208,10 +205,10 @@
      */
     @Test
     public void testUserSpecifiedCamelContext() throws Exception {
-        final AtomicInteger cnt = new AtomicInteger();
 
-        // Create a CamelContext with a probe that'll help us know if it has been used.
-        CamelContext context = new DefaultCamelContext();
+        CamelContext context = streamer.getCamelContext();
+
+        final AtomicInteger cnt = new AtomicInteger();
         context.setTracing(true);
         context.addLifecycleStrategy(new LifecycleStrategySupport() {
             @Override public void onEndpointAdd(Endpoint endpoint) {
@@ -220,7 +217,6 @@
         });
 
         streamer.setSingleTupleExtractor(singleTupleExtractor());
-        streamer.setCamelContext(context);
 
         // Subscribe to cache PUT events.
         CountDownLatch latch = subscribeToPutEvents(50);
@@ -232,7 +228,7 @@
         sendMessages(0, 50, false);
 
         // Assertions.
-        assertTrue(latch.await(10, TimeUnit.SECONDS));
+        assertTrue(latch.await(100, TimeUnit.SECONDS));
         assertCacheEntriesLoaded(50);
         assertTrue(cnt.get() > 0);
     }
@@ -247,7 +243,7 @@
 
         PropertiesComponent pc = new PropertiesComponent("camel.test.properties");
 
-        context.addComponent("properties", pc);
+        context.setPropertiesComponent(pc);
 
         // Replace the context path in the test URL with the property placeholder.
         url = url.replaceAll("/ignite", "{{test.contextPath}}");
@@ -258,6 +254,9 @@
         streamer.setSingleTupleExtractor(singleTupleExtractor());
         streamer.setCamelContext(context);
 
+        producerTemplate = new DefaultProducerTemplate(context);
+        producerTemplate.start();
+
         // Subscribe to cache PUT events.
         CountDownLatch latch = subscribeToPutEvents(50);
 
@@ -290,8 +289,11 @@
             fail("Streamer started; should have failed.");
         }
         catch (IgniteException ignored) {
-            assertTrue(streamer.getCamelContext().getStatus() == ServiceStatus.Stopped);
-            assertTrue(streamer.getCamelContext().getEndpointRegistry().size() == 0);
+            CamelContext context = streamer.getCamelContext();
+            ServiceStatus status = context.getStatus();
+            assertEquals("streamer should be in a stopped state, instead is " + status, status, ServiceStatus.Stopped);
+            assertEquals("endpoint registry has > 0 elements, size: " + context.getEndpointRegistry().size(),
+                context.getEndpointRegistry().size(), 0);
         }
     }
 
@@ -312,10 +314,10 @@
     }
 
     /**
-     * @throws IOException
+     * @throws Exception
      * @return HTTP response payloads.
      */
-    private List<String> sendMessages(int fromIdx, int cnt, boolean singleMessage) throws IOException {
+    private List<String> sendMessages(int fromIdx, int cnt, boolean singleMessage) throws Exception {
         List<String> responses = Lists.newArrayList();
 
         if (singleMessage) {
@@ -324,27 +326,23 @@
             for (int i = fromIdx; i < fromIdx + cnt; i++)
                 sb.append(i).append(",").append(TEST_DATA.get(i)).append("\n");
 
-            Request request = new Request.Builder()
-                .url(url)
-                .post(RequestBody.create(TEXT_PLAIN, sb.toString()))
-                .build();
+            Exchange out = producerTemplate.send(url, exchange -> exchange.getIn().setBody(sb));
 
-            Response response = httpClient.newCall(request).execute();
+            if (out.getException() != null)
+                throw out.getException();
 
-            responses.add(response.body().string());
+            responses.add(out.getMessage().toString());
         }
         else {
             for (int i = fromIdx; i < fromIdx + cnt; i++) {
                 String payload = i + "," + TEST_DATA.get(i);
 
-                Request request = new Request.Builder()
-                    .url(url)
-                    .post(RequestBody.create(TEXT_PLAIN, payload))
-                    .build();
+                Exchange out = producerTemplate.send(url, exchange -> exchange.getIn().setBody(payload));
 
-                Response response = httpClient.newCall(request).execute();
+                if (out.getException() != null)
+                    throw out.getException();
 
-                responses.add(response.body().string());
+                responses.add(out.getMessage().getBody(String.class));
             }
         }
 
diff --git a/parent/pom.xml b/parent/pom.xml
index 9610e2a..6444b27 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -55,7 +55,7 @@
         <aspectj.version>1.8.13</aspectj.version>
         <aws.sdk.bundle.version>1.10.12_1</aws.sdk.bundle.version>
         <aws.sdk.version>1.11.75</aws.sdk.version>
-        <camel.version>2.22.0</camel.version>
+        <camel.version>3.4.3</camel.version>
         <aws.encryption.sdk.version>1.3.2</aws.encryption.sdk.version>
         <commons.beanutils.bundle.version>1.9.2_1</commons.beanutils.bundle.version>
         <commons.beanutils.version>1.9.3</commons.beanutils.version>