Croway commented on code in PR #1859: URL: https://github.com/apache/camel-kafka-connector/pull/1859#discussion_r4471414257
########## docs/modules/ROOT/pages/user-guide/webhook-source.adoc: ########## @@ -0,0 +1,144 @@ +[[WebhookSource-WebhookSource]] += Webhook Source Connector + +The `camel-webhook-source-kafka-connector` receives HTTP requests and writes them to Kafka. It is generated from the +https://camel.apache.org/camel-kamelets/latest/webhook-source.html[Webhook Source Kamelet], which consumes from the +https://camel.apache.org/components/latest/platform-http-component.html[Platform HTTP] component, and it ships the +embedded HTTP server of Camel Main, +https://camel.apache.org/components/latest/others/platform-http-main.html[camel-platform-http-main], which is based on +Vert.x. The server runs inside the Kafka Connect worker that runs the task of the connector. + +[#webhook-enabling-the-http-server] +== Enabling the HTTP server + +The embedded HTTP server is disabled by default. The connector passes its configuration to Camel Main, so the server +is enabled and configured with the `camel.server.*` options of +https://camel.apache.org/components/latest/others/main.html[Camel Main], set in the connector configuration: + +[width="100%",cols="2,5,^1",options="header"] +|=== +| Name | Description | Default +| camel.server.enabled | Starts the embedded HTTP server. The connector does not start without it. | false +| camel.server.host | The address the server binds to. | 0.0.0.0 +| camel.server.port | The port the server listens on. | 8080 +| camel.server.path | The context path, which prefixes the path of the connector. | / +| camel.kamelet.webhook-source.subpath | The path where the connector receives requests. | webhook +| camel.server.maxBodySize | The maximum size, in bytes, of a request body. | +|=== + +The value of a record is the request body, as a `byte[]`. The `ByteArrayConverter` writes it to the topic unchanged. Review Comment: This only holds once #1852 is merged. On this branch `CamelSourceTask` puts the body in the record as it is, and for platform-http-vertx that's a Vert.x `Buffer` (a `StreamCache` with stream caching), not a `byte[]`. With `value.converter=ByteArrayConverter`, `fromConnectData` throws a `DataException` and the task fails. That's also the "Java object reference" the `StringConverter` sentence below refers to. Even with #1852, `application/x-www-form-urlencoded` and multipart requests (Slack, Twilio, ...) don't become a `byte[]`: `populateMultiFormData` sets the body to a `HashMap` of the form fields. Should this PR wait for #1852, and the page mention form bodies? ########## connectors/camel-kafka-connector-kamelet-fix-dependencies.properties: ########## @@ -27,4 +27,8 @@ global=org.apache.camel.kafkaconnector:camel-kafka-connector,org.apache.camel:ca #camel-sjms2=org.apache.activemq:activemq-client::runtime,org.apache.activemq:artemis-jms-client::runtime #exclude_camel-sjms2= #additional_properties_camel-sjms2=camel.component.sjms2.connection-factory=#class:org.apache.activemq.ActiveMQConnectionFactory,camel.component.sjms2.connection-factory.brokerURL=tcp://localhost:61616 -camel-elasticsearch-index-sink=org.apache.camel:camel-bean \ No newline at end of file +camel-elasticsearch-index-sink=org.apache.camel:camel-bean + +# the webhook-source kamelet only declares camel-platform-http, which has no HTTP server of its own: ship the +# embedded Camel Main HTTP server (Vert.x), started by setting camel.server.enabled=true in the connector config +camel-webhook-source=org.apache.camel:camel-platform-http-main Review Comment: `VertxPlatformHttpServer` swallows a failed bind: the `RuntimeException` is thrown inside the Vert.x `listen` callback, so `startServer()` returns normally and the task reports RUNNING with no listener. The page documents it, but it bites in more cases than two connectors on one port: - `tasks.max > 1`: every task after the first one on a worker listens on nothing. - Restart or config update with requests in flight: the old task's `stop()` can take longer than `task.shutdown.graceful.timeout.ms` (5s by default) while the exchanges wait for their reply (up to 30s). Connect then starts the new task on the same worker while the old server still holds the port, the new bind fails, and once the old server closes nothing listens on the port until the next manual restart. Could the task fail `start()` when the server has no bound port, and/or could the connector cap `taskConfigs` to 1? ########## connectors/camel-kafka-connector-kamelet-fix-dependencies.properties: ########## @@ -27,4 +27,8 @@ global=org.apache.camel.kafkaconnector:camel-kafka-connector,org.apache.camel:ca #camel-sjms2=org.apache.activemq:activemq-client::runtime,org.apache.activemq:artemis-jms-client::runtime #exclude_camel-sjms2= #additional_properties_camel-sjms2=camel.component.sjms2.connection-factory=#class:org.apache.activemq.ActiveMQConnectionFactory,camel.component.sjms2.connection-factory.brokerURL=tcp://localhost:61616 -camel-elasticsearch-index-sink=org.apache.camel:camel-bean \ No newline at end of file +camel-elasticsearch-index-sink=org.apache.camel:camel-bean + +# the webhook-source kamelet only declares camel-platform-http, which has no HTTP server of its own: ship the +# embedded Camel Main HTTP server (Vert.x), started by setting camel.server.enabled=true in the connector config Review Comment: With the server disabled by default, a fresh install fails at task start with `No PlatformHttpEngine created and setup in registry with name: platform-http-router-8080`, which doesn't point at `camel.server.enabled`. The generated `CamelWebhooksourceSourceConnector.properties` example and the generated reference page don't mention it either, only the new user-guide page does. Since the connector can't work without the server, what about `additional_properties_camel-webhook-source=camel.server.enabled=true`, so it ends up in the task defaults (`getDefaultConfig()`)? Otherwise, at least a fail-fast with a message that names the option. ########## tests/itests-webhook/src/test/java/org/apache/camel/kafkaconnector/webhook/source/CamelSourceWebhookITCase.java: ########## @@ -0,0 +1,117 @@ +/* + * 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.camel.kafkaconnector.webhook.source; + +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.time.Duration; + +import org.apache.camel.kafkaconnector.common.ConnectorPropertyFactory; +import org.apache.camel.kafkaconnector.common.test.CamelSourceTestSupport; +import org.apache.camel.kafkaconnector.common.test.TestMessageConsumer; +import org.apache.camel.kafkaconnector.common.utils.NetworkUtils; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.header.Header; +import org.awaitility.Awaitility; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.junit.jupiter.api.Timeout; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +/** + * Starts the webhook source connector with the embedded HTTP server enabled, sends a request to the path the Kamelet + * exposes and checks that its body lands on the Kafka topic. + */ +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public class CamelSourceWebhookITCase extends CamelSourceTestSupport { + private static final Logger LOG = LoggerFactory.getLogger(CamelSourceWebhookITCase.class); + private static final String HTTP_HOST = "127.0.0.1"; + private static final int HTTP_PORT = NetworkUtils.getFreePort(HTTP_HOST); Review Comment: On Linux `NetworkUtils.getFreePort(host)` starts at 49152, inside the kernel's ephemeral range, and the port is only bound once the task starts. In between, an outgoing Kafka connection can take it, or a later `getFreePort()` (e.g. for the Connect REST listener) can return it again, since `freeStartingPort` only advances when the first candidate was free. The bind failure is swallowed, so the test only fails after the full 60s Awaitility wait. The netty-http IT uses an explicit range, `NetworkUtils.getFreePort("localhost", 30000, 40000)`; I'd do the same here. ########## tests/itests-webhook/src/test/java/org/apache/camel/kafkaconnector/webhook/source/CamelWebhookPropertyFactory.java: ########## @@ -0,0 +1,56 @@ +/* + * 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.camel.kafkaconnector.webhook.source; + +import org.apache.camel.kafkaconnector.common.SourceConnectorPropertyFactory; + +final class CamelWebhookPropertyFactory extends SourceConnectorPropertyFactory<CamelWebhookPropertyFactory> { + private CamelWebhookPropertyFactory() { + + } + + public CamelWebhookPropertyFactory withSubpath(String subpath) { + return setProperty("camel.kamelet.webhook-source.subpath", subpath); + } + + public CamelWebhookPropertyFactory withServerEnabled(boolean enabled) { + return setProperty("camel.server.enabled", String.valueOf(enabled)); + } + + public CamelWebhookPropertyFactory withServerHost(String host) { + return setProperty("camel.server.host", host); + } + + public CamelWebhookPropertyFactory withServerPort(int port) { + return setProperty("camel.server.port", String.valueOf(port)); + } + + public CamelWebhookPropertyFactory withCamelTypeConverterTransformTo(String targetClass) { Review Comment: nit: `BasicConnectorPropertyFactory` already has `withTransformsConfig(name).withEntry(...).end()` (used by the slack and ssh ITs), so this doesn't need another hand-written copy of the transform keys. ########## docs/modules/ROOT/pages/user-guide/webhook-source.adoc: ########## @@ -0,0 +1,144 @@ +[[WebhookSource-WebhookSource]] += Webhook Source Connector + +The `camel-webhook-source-kafka-connector` receives HTTP requests and writes them to Kafka. It is generated from the +https://camel.apache.org/camel-kamelets/latest/webhook-source.html[Webhook Source Kamelet], which consumes from the +https://camel.apache.org/components/latest/platform-http-component.html[Platform HTTP] component, and it ships the +embedded HTTP server of Camel Main, +https://camel.apache.org/components/latest/others/platform-http-main.html[camel-platform-http-main], which is based on +Vert.x. The server runs inside the Kafka Connect worker that runs the task of the connector. + +[#webhook-enabling-the-http-server] +== Enabling the HTTP server + +The embedded HTTP server is disabled by default. The connector passes its configuration to Camel Main, so the server +is enabled and configured with the `camel.server.*` options of +https://camel.apache.org/components/latest/others/main.html[Camel Main], set in the connector configuration: + +[width="100%",cols="2,5,^1",options="header"] +|=== +| Name | Description | Default +| camel.server.enabled | Starts the embedded HTTP server. The connector does not start without it. | false +| camel.server.host | The address the server binds to. | 0.0.0.0 +| camel.server.port | The port the server listens on. | 8080 +| camel.server.path | The context path, which prefixes the path of the connector. | / +| camel.kamelet.webhook-source.subpath | The path where the connector receives requests. | webhook +| camel.server.maxBodySize | The maximum size, in bytes, of a request body. | +|=== + +The value of a record is the request body, as a `byte[]`. The `ByteArrayConverter` writes it to the topic unchanged. +To get a string value, convert it with the `CamelTypeConverterTransform` SMT, as in the configuration below: a +`StringConverter` alone writes a Java object reference instead of the body. A complete configuration: + +[source,properties] +---- +name=webhook-source +connector.class=org.apache.camel.kafkaconnector.webhooksource.CamelWebhooksourceSourceConnector +tasks.max=1 +topics=webhook-events + +key.converter=org.apache.kafka.connect.storage.StringConverter +value.converter=org.apache.kafka.connect.storage.StringConverter + +transforms=tostring +transforms.tostring.type=org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform$Value +transforms.tostring.target.type=java.lang.String + +camel.server.enabled=true +camel.server.port=8080 +camel.kamelet.webhook-source.subpath=webhook +---- + +The connector then receives requests on `http://<worker>:8080/webhook`: + +[source,bash] +---- +curl -X POST -H 'Content-Type: application/json' -d '{"event":"push"}' http://<worker>:8080/webhook +---- + +Each request becomes one record. The request headers, and request details such as the method and the path +(`CamelHttpMethod`, `CamelHttpPath`), become record headers with the `CamelHeader.` prefix, unless +`camel.map.headers=false` is set. + +The response is sent once Kafka Connect has written the record to the topic, so a `200` response means that the +record is in Kafka. If that takes more than 30 seconds, the client receives a `500` response. The record can still be +written afterwards, so a client that retries can produce duplicates. + +[#webhook-deployment] +== Deployment + +* *One task, one port.* Every task starts its own HTTP server on `camel.server.port`. Set `tasks.max=1`, and give each +webhook connector of the Kafka Connect cluster its own port. Only the first server binds a port on a worker: another +one logs `Failed to start Vert.x HttpServer on 0.0.0.0:8080, reason: Address already in use`, and its task keeps +running without receiving any request. Review Comment: It's worse than not receiving requests: if the two connectors also share the subpath (e.g. the default `webhook`), the requests meant for the second connector are served by the router of the first one, get a `200`, and land in the first connector's topic. Worth spelling out, since it's silent misrouting rather than a dead listener. ########## docs/modules/ROOT/pages/user-guide/webhook-source.adoc: ########## @@ -0,0 +1,144 @@ +[[WebhookSource-WebhookSource]] += Webhook Source Connector + +The `camel-webhook-source-kafka-connector` receives HTTP requests and writes them to Kafka. It is generated from the +https://camel.apache.org/camel-kamelets/latest/webhook-source.html[Webhook Source Kamelet], which consumes from the +https://camel.apache.org/components/latest/platform-http-component.html[Platform HTTP] component, and it ships the +embedded HTTP server of Camel Main, +https://camel.apache.org/components/latest/others/platform-http-main.html[camel-platform-http-main], which is based on +Vert.x. The server runs inside the Kafka Connect worker that runs the task of the connector. + +[#webhook-enabling-the-http-server] +== Enabling the HTTP server + +The embedded HTTP server is disabled by default. The connector passes its configuration to Camel Main, so the server +is enabled and configured with the `camel.server.*` options of +https://camel.apache.org/components/latest/others/main.html[Camel Main], set in the connector configuration: + +[width="100%",cols="2,5,^1",options="header"] +|=== +| Name | Description | Default +| camel.server.enabled | Starts the embedded HTTP server. The connector does not start without it. | false +| camel.server.host | The address the server binds to. | 0.0.0.0 +| camel.server.port | The port the server listens on. | 8080 +| camel.server.path | The context path, which prefixes the path of the connector. | / +| camel.kamelet.webhook-source.subpath | The path where the connector receives requests. | webhook +| camel.server.maxBodySize | The maximum size, in bytes, of a request body. | +|=== + +The value of a record is the request body, as a `byte[]`. The `ByteArrayConverter` writes it to the topic unchanged. +To get a string value, convert it with the `CamelTypeConverterTransform` SMT, as in the configuration below: a +`StringConverter` alone writes a Java object reference instead of the body. A complete configuration: + +[source,properties] +---- +name=webhook-source +connector.class=org.apache.camel.kafkaconnector.webhooksource.CamelWebhooksourceSourceConnector +tasks.max=1 +topics=webhook-events + +key.converter=org.apache.kafka.connect.storage.StringConverter +value.converter=org.apache.kafka.connect.storage.StringConverter + +transforms=tostring +transforms.tostring.type=org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform$Value +transforms.tostring.target.type=java.lang.String + +camel.server.enabled=true +camel.server.port=8080 +camel.kamelet.webhook-source.subpath=webhook +---- + +The connector then receives requests on `http://<worker>:8080/webhook`: + +[source,bash] +---- +curl -X POST -H 'Content-Type: application/json' -d '{"event":"push"}' http://<worker>:8080/webhook +---- + +Each request becomes one record. The request headers, and request details such as the method and the path +(`CamelHttpMethod`, `CamelHttpPath`), become record headers with the `CamelHeader.` prefix, unless +`camel.map.headers=false` is set. + +The response is sent once Kafka Connect has written the record to the topic, so a `200` response means that the +record is in Kafka. If that takes more than 30 seconds, the client receives a `500` response. The record can still be +written afterwards, so a client that retries can produce duplicates. + +[#webhook-deployment] +== Deployment + +* *One task, one port.* Every task starts its own HTTP server on `camel.server.port`. Set `tasks.max=1`, and give each +webhook connector of the Kafka Connect cluster its own port. Only the first server binds a port on a worker: another +one logs `Failed to start Vert.x HttpServer on 0.0.0.0:8080, reason: Address already in use`, and its task keeps +running without receiving any request. +* *The listener moves with the task.* The server runs only on the worker that runs the task, and moves to another +worker when Kafka Connect reassigns the task, for example when a worker restarts or fails, when workers are added, or +when the connector is updated. The `worker_id` of the task, in the response of `GET /connectors/<name>/status` of the +Kafka Connect REST API, tells which worker it is. Make the port reachable on every worker that can run the task (for +example, as a container port) and send the requests to the worker that runs it. A load balancer, or a Kubernetes +`Service`, in front of several workers also forwards requests to workers where nothing listens on the port, and those +requests fail. Running the connector on a Kafka Connect cluster with a single worker avoids this. +* *No clashes with other listeners.* The port must differ from the other ports of the worker, such as the one of the +Kafka Connect REST API (8083 by default). + +[#webhook-security] +== Security + +The HTTP server accepts requests from any client that can reach the port, and writes every accepted request to Kafka. +With the defaults, it binds to every network interface, without authentication and without TLS. In the terms of the +xref:security-model.adoc[security model], the clients are an external system: what they send is untrusted data. + +* *Restrict who can reach the port.* Bind `camel.server.host` to an internal address, and use firewall rules or +network policies so that only the systems that call the webhook can reach it. +* *Require authentication.* Set `camel.server.authenticationEnabled=true` together with one of the mechanisms below. +With `camel.server.authenticationEnabled=true` alone, the server only logs a warning and starts *without* +authentication. `camel.server.authenticationPath` selects the protected paths, all of them (`/*`) by default. +** HTTP basic authentication: `camel.server.basicPropertiesFile`, the path of a Vert.x properties file on the worker, +with entries such as `user.<name>=<password>,<role>` and `role.<role>=*`. The passwords are stored in clear text, so +restrict the access to the file. +** JWT bearer tokens: `camel.server.jwtKeystoreType`, `camel.server.jwtKeystorePath`, +`camel.server.jwtKeystorePassword`, and `camel.server.jwtIssuer` and/or `camel.server.jwtAudience`. +* *Use TLS.* Configure the global SSL options of Camel Main in the connector configuration, for example +`camel.ssl.enabled=true`, `camel.ssl.keyStore=file:/path/to/keystore.p12` and `camel.ssl.keystorePassword`: the +server then uses HTTPS. +* *Keep secrets out of the connector configuration.* Resolve the passwords with a Kafka Connect config provider, for +example `camel.ssl.keystorePassword=${file:/opt/kafka/secrets/webhook.properties:keystore.password}` with the +`FileConfigProvider` configured on the workers. +* *Limit the size of the requests* with `camel.server.maxBodySize`. Review Comment: The server also accepts multipart file uploads by default (`camel.server.fileUploadEnabled=true`) and writes them to the worker's disk, from any client when authentication is off. A webhook doesn't need them, so I'd recommend `camel.server.fileUploadEnabled=false` next to `maxBodySize`. ########## docs/modules/ROOT/pages/user-guide/webhook-source.adoc: ########## @@ -0,0 +1,144 @@ +[[WebhookSource-WebhookSource]] += Webhook Source Connector + +The `camel-webhook-source-kafka-connector` receives HTTP requests and writes them to Kafka. It is generated from the +https://camel.apache.org/camel-kamelets/latest/webhook-source.html[Webhook Source Kamelet], which consumes from the +https://camel.apache.org/components/latest/platform-http-component.html[Platform HTTP] component, and it ships the +embedded HTTP server of Camel Main, +https://camel.apache.org/components/latest/others/platform-http-main.html[camel-platform-http-main], which is based on +Vert.x. The server runs inside the Kafka Connect worker that runs the task of the connector. + +[#webhook-enabling-the-http-server] +== Enabling the HTTP server + +The embedded HTTP server is disabled by default. The connector passes its configuration to Camel Main, so the server +is enabled and configured with the `camel.server.*` options of +https://camel.apache.org/components/latest/others/main.html[Camel Main], set in the connector configuration: + +[width="100%",cols="2,5,^1",options="header"] +|=== +| Name | Description | Default +| camel.server.enabled | Starts the embedded HTTP server. The connector does not start without it. | false +| camel.server.host | The address the server binds to. | 0.0.0.0 +| camel.server.port | The port the server listens on. | 8080 +| camel.server.path | The context path, which prefixes the path of the connector. | / +| camel.kamelet.webhook-source.subpath | The path where the connector receives requests. | webhook +| camel.server.maxBodySize | The maximum size, in bytes, of a request body. | +|=== + +The value of a record is the request body, as a `byte[]`. The `ByteArrayConverter` writes it to the topic unchanged. +To get a string value, convert it with the `CamelTypeConverterTransform` SMT, as in the configuration below: a +`StringConverter` alone writes a Java object reference instead of the body. A complete configuration: + +[source,properties] +---- +name=webhook-source +connector.class=org.apache.camel.kafkaconnector.webhooksource.CamelWebhooksourceSourceConnector +tasks.max=1 +topics=webhook-events + +key.converter=org.apache.kafka.connect.storage.StringConverter +value.converter=org.apache.kafka.connect.storage.StringConverter + +transforms=tostring +transforms.tostring.type=org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform$Value +transforms.tostring.target.type=java.lang.String + +camel.server.enabled=true +camel.server.port=8080 +camel.kamelet.webhook-source.subpath=webhook +---- + +The connector then receives requests on `http://<worker>:8080/webhook`: + +[source,bash] +---- +curl -X POST -H 'Content-Type: application/json' -d '{"event":"push"}' http://<worker>:8080/webhook +---- + +Each request becomes one record. The request headers, and request details such as the method and the path +(`CamelHttpMethod`, `CamelHttpPath`), become record headers with the `CamelHeader.` prefix, unless +`camel.map.headers=false` is set. + +The response is sent once Kafka Connect has written the record to the topic, so a `200` response means that the +record is in Kafka. If that takes more than 30 seconds, the client receives a `500` response. The record can still be +written afterwards, so a client that retries can produce duplicates. + +[#webhook-deployment] +== Deployment + +* *One task, one port.* Every task starts its own HTTP server on `camel.server.port`. Set `tasks.max=1`, and give each +webhook connector of the Kafka Connect cluster its own port. Only the first server binds a port on a worker: another +one logs `Failed to start Vert.x HttpServer on 0.0.0.0:8080, reason: Address already in use`, and its task keeps +running without receiving any request. +* *The listener moves with the task.* The server runs only on the worker that runs the task, and moves to another +worker when Kafka Connect reassigns the task, for example when a worker restarts or fails, when workers are added, or +when the connector is updated. The `worker_id` of the task, in the response of `GET /connectors/<name>/status` of the +Kafka Connect REST API, tells which worker it is. Make the port reachable on every worker that can run the task (for +example, as a container port) and send the requests to the worker that runs it. A load balancer, or a Kubernetes +`Service`, in front of several workers also forwards requests to workers where nothing listens on the port, and those +requests fail. Running the connector on a Kafka Connect cluster with a single worker avoids this. +* *No clashes with other listeners.* The port must differ from the other ports of the worker, such as the one of the +Kafka Connect REST API (8083 by default). + +[#webhook-security] +== Security + +The HTTP server accepts requests from any client that can reach the port, and writes every accepted request to Kafka. +With the defaults, it binds to every network interface, without authentication and without TLS. In the terms of the +xref:security-model.adoc[security model], the clients are an external system: what they send is untrusted data. + +* *Restrict who can reach the port.* Bind `camel.server.host` to an internal address, and use firewall rules or +network policies so that only the systems that call the webhook can reach it. +* *Require authentication.* Set `camel.server.authenticationEnabled=true` together with one of the mechanisms below. +With `camel.server.authenticationEnabled=true` alone, the server only logs a warning and starts *without* +authentication. `camel.server.authenticationPath` selects the protected paths, all of them (`/*`) by default. +** HTTP basic authentication: `camel.server.basicPropertiesFile`, the path of a Vert.x properties file on the worker, +with entries such as `user.<name>=<password>,<role>` and `role.<role>=*`. The passwords are stored in clear text, so +restrict the access to the file. +** JWT bearer tokens: `camel.server.jwtKeystoreType`, `camel.server.jwtKeystorePath`, +`camel.server.jwtKeystorePassword`, and `camel.server.jwtIssuer` and/or `camel.server.jwtAudience`. +* *Use TLS.* Configure the global SSL options of Camel Main in the connector configuration, for example +`camel.ssl.enabled=true`, `camel.ssl.keyStore=file:/path/to/keystore.p12` and `camel.ssl.keystorePassword`: the +server then uses HTTPS. +* *Keep secrets out of the connector configuration.* Resolve the passwords with a Kafka Connect config provider, for +example `camel.ssl.keystorePassword=${file:/opt/kafka/secrets/webhook.properties:keystore.password}` with the +`FileConfigProvider` configured on the workers. +* *Limit the size of the requests* with `camel.server.maxBodySize`. +* *Do not trust the record headers.* When a Camel sink connector consumes the topic, set `camel.remove.headers.pattern` +on it, so that the request headers do not become Camel headers there, or set `camel.map.headers=false` on the webhook +connector when the headers are not needed. See the deployment hardening section of the +xref:security-model.adoc[security model]. + +[#webhook-troubleshooting] +== Troubleshooting + +`No platform-http-engine service could be found in the classpath. Add camel-platform-http-vertx to classpath.`:: +The connector package does not contain the HTTP server, as is the case for the packages up to 4.22.0. For 4.22.0, add +the jars of `camel-platform-http-main` and of its dependencies to the connector directory: Review Comment: Worth adding that the worker has to be restarted afterwards: Kafka Connect scans `plugin.path` only at startup, so restarting the connector or the task through the REST API keeps failing with the same error, and the workaround looks broken. ########## docs/modules/ROOT/pages/user-guide/webhook-source.adoc: ########## @@ -0,0 +1,144 @@ +[[WebhookSource-WebhookSource]] += Webhook Source Connector + +The `camel-webhook-source-kafka-connector` receives HTTP requests and writes them to Kafka. It is generated from the +https://camel.apache.org/camel-kamelets/latest/webhook-source.html[Webhook Source Kamelet], which consumes from the +https://camel.apache.org/components/latest/platform-http-component.html[Platform HTTP] component, and it ships the +embedded HTTP server of Camel Main, +https://camel.apache.org/components/latest/others/platform-http-main.html[camel-platform-http-main], which is based on +Vert.x. The server runs inside the Kafka Connect worker that runs the task of the connector. + +[#webhook-enabling-the-http-server] +== Enabling the HTTP server + +The embedded HTTP server is disabled by default. The connector passes its configuration to Camel Main, so the server +is enabled and configured with the `camel.server.*` options of +https://camel.apache.org/components/latest/others/main.html[Camel Main], set in the connector configuration: + +[width="100%",cols="2,5,^1",options="header"] +|=== +| Name | Description | Default +| camel.server.enabled | Starts the embedded HTTP server. The connector does not start without it. | false +| camel.server.host | The address the server binds to. | 0.0.0.0 +| camel.server.port | The port the server listens on. | 8080 +| camel.server.path | The context path, which prefixes the path of the connector. | / +| camel.kamelet.webhook-source.subpath | The path where the connector receives requests. | webhook +| camel.server.maxBodySize | The maximum size, in bytes, of a request body. | +|=== + +The value of a record is the request body, as a `byte[]`. The `ByteArrayConverter` writes it to the topic unchanged. +To get a string value, convert it with the `CamelTypeConverterTransform` SMT, as in the configuration below: a +`StringConverter` alone writes a Java object reference instead of the body. A complete configuration: + +[source,properties] +---- +name=webhook-source +connector.class=org.apache.camel.kafkaconnector.webhooksource.CamelWebhooksourceSourceConnector +tasks.max=1 +topics=webhook-events + +key.converter=org.apache.kafka.connect.storage.StringConverter +value.converter=org.apache.kafka.connect.storage.StringConverter + +transforms=tostring +transforms.tostring.type=org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform$Value +transforms.tostring.target.type=java.lang.String + +camel.server.enabled=true +camel.server.port=8080 +camel.kamelet.webhook-source.subpath=webhook +---- + +The connector then receives requests on `http://<worker>:8080/webhook`: + +[source,bash] +---- +curl -X POST -H 'Content-Type: application/json' -d '{"event":"push"}' http://<worker>:8080/webhook +---- + +Each request becomes one record. The request headers, and request details such as the method and the path +(`CamelHttpMethod`, `CamelHttpPath`), become record headers with the `CamelHeader.` prefix, unless +`camel.map.headers=false` is set. + +The response is sent once Kafka Connect has written the record to the topic, so a `200` response means that the Review Comment: Not quite: Kafka Connect also calls `commitRecord(record, null)` for records it drops, i.e. filtered by an SMT or predicate, or failed conversion/send with `errors.tolerance=all`. `CamelSourceTask.commitRecord` ignores the metadata and completes the exchange, so the client gets a `200` for a record that never reached the topic. I'd soften this to "a `200` means Kafka Connect is done with the record" and list the cases where it is dropped. ########## tests/itests-webhook/src/test/java/org/apache/camel/kafkaconnector/webhook/source/CamelSourceWebhookITCase.java: ########## @@ -0,0 +1,117 @@ +/* + * 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.camel.kafkaconnector.webhook.source; + +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.time.Duration; + +import org.apache.camel.kafkaconnector.common.ConnectorPropertyFactory; +import org.apache.camel.kafkaconnector.common.test.CamelSourceTestSupport; +import org.apache.camel.kafkaconnector.common.test.TestMessageConsumer; +import org.apache.camel.kafkaconnector.common.utils.NetworkUtils; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.header.Header; +import org.awaitility.Awaitility; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.junit.jupiter.api.Timeout; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +/** + * Starts the webhook source connector with the embedded HTTP server enabled, sends a request to the path the Kamelet + * exposes and checks that its body lands on the Kafka topic. + */ +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public class CamelSourceWebhookITCase extends CamelSourceTestSupport { + private static final Logger LOG = LoggerFactory.getLogger(CamelSourceWebhookITCase.class); + private static final String HTTP_HOST = "127.0.0.1"; + private static final int HTTP_PORT = NetworkUtils.getFreePort(HTTP_HOST); + private static final String SUBPATH = "ckc-webhook"; + private static final String TEST_MESSAGE = "{\"event\":\"ckc-webhook-test\"}"; + + private final HttpClient httpClient = HttpClient.newHttpClient(); + private final int expect = 1; + private String topicName; + + @Override + protected String[] getConnectorsInTest() { + return new String[] {"camel-webhook-source-kafka-connector"}; + } + + @BeforeEach + public void setUp() { + topicName = getTopicForTest(this); + } + + @Test + @Timeout(90) Review Comment: 90s doesn't cover the waits inside the test: up to 30s in `initializeConnectorBlocking`, then up to 60s of Awaitility (30s per request), then the consumer. On a slow agent JUnit kills the test first and reports a bare timeout instead of the Awaitility or assertion failure. I'd raise it, or drop it and rely on the inner timeouts. ########## docs/modules/ROOT/pages/user-guide/webhook-source.adoc: ########## @@ -0,0 +1,144 @@ +[[WebhookSource-WebhookSource]] += Webhook Source Connector + +The `camel-webhook-source-kafka-connector` receives HTTP requests and writes them to Kafka. It is generated from the +https://camel.apache.org/camel-kamelets/latest/webhook-source.html[Webhook Source Kamelet], which consumes from the +https://camel.apache.org/components/latest/platform-http-component.html[Platform HTTP] component, and it ships the +embedded HTTP server of Camel Main, +https://camel.apache.org/components/latest/others/platform-http-main.html[camel-platform-http-main], which is based on +Vert.x. The server runs inside the Kafka Connect worker that runs the task of the connector. + +[#webhook-enabling-the-http-server] +== Enabling the HTTP server + +The embedded HTTP server is disabled by default. The connector passes its configuration to Camel Main, so the server +is enabled and configured with the `camel.server.*` options of +https://camel.apache.org/components/latest/others/main.html[Camel Main], set in the connector configuration: + +[width="100%",cols="2,5,^1",options="header"] +|=== +| Name | Description | Default +| camel.server.enabled | Starts the embedded HTTP server. The connector does not start without it. | false +| camel.server.host | The address the server binds to. | 0.0.0.0 +| camel.server.port | The port the server listens on. | 8080 +| camel.server.path | The context path, which prefixes the path of the connector. | / +| camel.kamelet.webhook-source.subpath | The path where the connector receives requests. | webhook +| camel.server.maxBodySize | The maximum size, in bytes, of a request body. | +|=== + +The value of a record is the request body, as a `byte[]`. The `ByteArrayConverter` writes it to the topic unchanged. +To get a string value, convert it with the `CamelTypeConverterTransform` SMT, as in the configuration below: a +`StringConverter` alone writes a Java object reference instead of the body. A complete configuration: + +[source,properties] +---- +name=webhook-source +connector.class=org.apache.camel.kafkaconnector.webhooksource.CamelWebhooksourceSourceConnector +tasks.max=1 +topics=webhook-events + +key.converter=org.apache.kafka.connect.storage.StringConverter +value.converter=org.apache.kafka.connect.storage.StringConverter + +transforms=tostring +transforms.tostring.type=org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform$Value +transforms.tostring.target.type=java.lang.String + +camel.server.enabled=true +camel.server.port=8080 +camel.kamelet.webhook-source.subpath=webhook +---- + +The connector then receives requests on `http://<worker>:8080/webhook`: + +[source,bash] +---- +curl -X POST -H 'Content-Type: application/json' -d '{"event":"push"}' http://<worker>:8080/webhook +---- + +Each request becomes one record. The request headers, and request details such as the method and the path +(`CamelHttpMethod`, `CamelHttpPath`), become record headers with the `CamelHeader.` prefix, unless +`camel.map.headers=false` is set. + +The response is sent once Kafka Connect has written the record to the topic, so a `200` response means that the +record is in Kafka. If that takes more than 30 seconds, the client receives a `500` response. The record can still be +written afterwards, so a client that retries can produce duplicates. + +[#webhook-deployment] +== Deployment + +* *One task, one port.* Every task starts its own HTTP server on `camel.server.port`. Set `tasks.max=1`, and give each +webhook connector of the Kafka Connect cluster its own port. Only the first server binds a port on a worker: another +one logs `Failed to start Vert.x HttpServer on 0.0.0.0:8080, reason: Address already in use`, and its task keeps +running without receiving any request. +* *The listener moves with the task.* The server runs only on the worker that runs the task, and moves to another +worker when Kafka Connect reassigns the task, for example when a worker restarts or fails, when workers are added, or +when the connector is updated. The `worker_id` of the task, in the response of `GET /connectors/<name>/status` of the +Kafka Connect REST API, tells which worker it is. Make the port reachable on every worker that can run the task (for +example, as a container port) and send the requests to the worker that runs it. A load balancer, or a Kubernetes +`Service`, in front of several workers also forwards requests to workers where nothing listens on the port, and those +requests fail. Running the connector on a Kafka Connect cluster with a single worker avoids this. +* *No clashes with other listeners.* The port must differ from the other ports of the worker, such as the one of the +Kafka Connect REST API (8083 by default). + +[#webhook-security] +== Security + +The HTTP server accepts requests from any client that can reach the port, and writes every accepted request to Kafka. +With the defaults, it binds to every network interface, without authentication and without TLS. In the terms of the +xref:security-model.adoc[security model], the clients are an external system: what they send is untrusted data. + +* *Restrict who can reach the port.* Bind `camel.server.host` to an internal address, and use firewall rules or +network policies so that only the systems that call the webhook can reach it. +* *Require authentication.* Set `camel.server.authenticationEnabled=true` together with one of the mechanisms below. +With `camel.server.authenticationEnabled=true` alone, the server only logs a warning and starts *without* +authentication. `camel.server.authenticationPath` selects the protected paths, all of them (`/*`) by default. +** HTTP basic authentication: `camel.server.basicPropertiesFile`, the path of a Vert.x properties file on the worker, +with entries such as `user.<name>=<password>,<role>` and `role.<role>=*`. The passwords are stored in clear text, so +restrict the access to the file. +** JWT bearer tokens: `camel.server.jwtKeystoreType`, `camel.server.jwtKeystorePath`, +`camel.server.jwtKeystorePassword`, and `camel.server.jwtIssuer` and/or `camel.server.jwtAudience`. +* *Use TLS.* Configure the global SSL options of Camel Main in the connector configuration, for example +`camel.ssl.enabled=true`, `camel.ssl.keyStore=file:/path/to/keystore.p12` and `camel.ssl.keystorePassword`: the +server then uses HTTPS. +* *Keep secrets out of the connector configuration.* Resolve the passwords with a Kafka Connect config provider, for +example `camel.ssl.keystorePassword=${file:/opt/kafka/secrets/webhook.properties:keystore.password}` with the +`FileConfigProvider` configured on the workers. +* *Limit the size of the requests* with `camel.server.maxBodySize`. +* *Do not trust the record headers.* When a Camel sink connector consumes the topic, set `camel.remove.headers.pattern` +on it, so that the request headers do not become Camel headers there, or set `camel.map.headers=false` on the webhook +connector when the headers are not needed. See the deployment hardening section of the +xref:security-model.adoc[security model]. + +[#webhook-troubleshooting] +== Troubleshooting + +`No platform-http-engine service could be found in the classpath. Add camel-platform-http-vertx to classpath.`:: Review Comment: With `camel.server.enabled=true`, as this page says to set, the 4.22.0 package fails earlier, in `BaseMainSupport.setHttpServerProperties`, with: ``` No platform-http-server service could be found in the classpath. Add camel-platform-http-main to classpath. ``` The `platform-http-engine` message only shows up without `camel.server.enabled`. Could you add this message here and to the Known Limitations entry in `troubleshooting.adoc`, so users searching for it find the workaround? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
