Add HTTP Sink (apache#17581)

deployment/terraform-ansible/deploy-pulsar.yaml

@@ -154,6 +154,7 @@
 #        - jdbc-mariadb
 #        - jdbc-postgres
 #        - jdbc-sqlite
+#        - http
         - kafka
 #        - kafka-connect-adaptor
 #        - kinesis

distribution/io/src/assemble/io.xml

@@ -47,6 +47,7 @@
     <file><source>${basedir}/../../pulsar-io/cassandra/target/pulsar-io-cassandra-${project.version}.nar</source></file>
     <file><source>${basedir}/../../pulsar-io/twitter/target/pulsar-io-twitter-${project.version}.nar</source></file>
     <file><source>${basedir}/../../pulsar-io/kafka/target/pulsar-io-kafka-${project.version}.nar</source></file>
+    <file><source>${basedir}/../../pulsar-io/http/target/pulsar-io-http-${project.version}.nar</source></file>
     <file><source>${basedir}/../../pulsar-io/kinesis/target/pulsar-io-kinesis-${project.version}.nar</source></file>
     <file><source>${basedir}/../../pulsar-io/rabbitmq/target/pulsar-io-rabbitmq-${project.version}.nar</source></file>
     <file><source>${basedir}/../../pulsar-io/nsq/target/pulsar-io-nsq-${project.version}.nar</source></file>

pom.xml

@@ -248,6 +248,7 @@ flexible messaging model and an intuitive client API.</description>
     <awaitility.version>4.2.0</awaitility.version>
     <reload4j.version>1.2.22</reload4j.version>
     <jettison.version>1.5.1</jettison.version>
+    <wiremock.version>2.33.2</wiremock.version>
 
     <!-- Plugin dependencies -->
     <protobuf-maven-plugin.version>0.6.1</protobuf-maven-plugin.version>

pulsar-io/docs/pom.xml

@@ -157,6 +157,11 @@
       <artifactId>pulsar-io-jdbc-openmldb</artifactId>
       <version>${project.version}</version>
     </dependency>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-io-http</artifactId>
+      <version>${project.version}</version>
+    </dependency>
     <dependency>
       <groupId>${project.groupId}</groupId>
       <artifactId>pulsar-io-kafka</artifactId>

pulsar-io/http/pom.xml

@@ -0,0 +1,87 @@
+<!--
+
+    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">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.pulsar</groupId>
+        <artifactId>pulsar-io</artifactId>
+        <version>2.11.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>pulsar-io-http</artifactId>
+    <name>Pulsar IO :: HTTP</name>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>pulsar-io-core</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-yaml</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.datatype</groupId>
+            <artifactId>jackson-datatype-jsr310</artifactId>
+            <version>${jackson.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.avro</groupId>
+            <artifactId>avro</artifactId>
+            <version>${avro.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.pulsar</groupId>
+            <artifactId>pulsar-client-original</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.github.tomakehurst</groupId>
+            <artifactId>wiremock-jre8</artifactId>
+            <version>${wiremock.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.nifi</groupId>
+                <artifactId>nifi-nar-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+</project>

pulsar-io/http/src/main/java/org/apache/pulsar/io/http/HttpSink.java

@@ -0,0 +1,130 @@
+/*
+ * 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.pulsar.io.http;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import java.io.IOException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.util.Base64;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.schema.GenericObject;
+import org.apache.pulsar.client.api.schema.KeyValueSchema;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.io.core.Sink;
+import org.apache.pulsar.io.core.SinkContext;
+
+/**
+ * A Sink that makes a POST request to a configured HTTP endpoint for each record (webhook).
+ * The body of the HTTP request is the JSON representation of the record value.
+ * Some headers are added to the HTTP request:
+ * <ul>
+ *   <li>PulsarTopic: the topic of the record</li>
+ *   <li>PulsarKey: the key of the record</li>
+ *   <li>PulsarEventTime: the event time of the record</li>
+ *   <li>PulsarPublishTime: the publish time of the record</li>
+ *   <li>PulsarMessageId: the ID of the message contained in the record</li>
+ *   <li>PulsarProperties-*: each record property is passed with the property name prefixed by PulsarProperties-</li>
+ * </ul>
+ */
+public class HttpSink implements Sink<GenericObject> {
+
+    HttpSinkConfig httpSinkConfig;
+    private HttpClient httpClient;
+    private ObjectMapper mapper;
+    private URI uri;
+
+    @Override
+    public void open(Map<String, Object> config, SinkContext sinkContext) throws Exception {
+        httpSinkConfig = HttpSinkConfig.load(config);
+        uri = new URI(httpSinkConfig.getUrl());
+        httpClient = HttpClient.newHttpClient();
+        mapper = new ObjectMapper().registerModule(new JavaTimeModule());
+    }
+
+    @Override
+    public void write(Record<GenericObject> record) throws Exception {
+        Object json = toJsonSerializable(record.getSchema(), record.getValue().getNativeObject());
+        byte[] bytes = mapper.writeValueAsBytes(json);
+        HttpRequest.Builder builder = HttpRequest.newBuilder()
+            .uri(uri)
+            .POST(HttpRequest.BodyPublishers.ofByteArray(bytes));
+        httpSinkConfig.getHeaders().forEach(builder::header);
+        record.getProperties().forEach((k, v) -> builder.header("PulsarProperties-" + k, v));
+        record.getTopicName().ifPresent(topic -> builder.header("PulsarTopic", topic));
+        record.getEventTime().ifPresent(eventTime -> builder.header("PulsarEventTime", eventTime.toString()));
+        record.getKey().ifPresent(key -> builder.header("PulsarKey", key));
+        record.getMessage().ifPresent(
+            message -> {
+              if (message.getMessageId() != null) {
+                String messageId = Base64.getEncoder().encodeToString(message.getMessageId().toByteArray());
+                builder.header("PulsarMessageId", messageId);
+              }
+              if (message.getPublishTime() != 0) {
+                builder.header("PulsarPublishTime", String.valueOf(message.getPublishTime()));
+              }
+            }
+        );
+        builder.header("Content-Type", "application/json");
+
+        HttpResponse<String> response = httpClient.send(builder.build(), HttpResponse.BodyHandlers.ofString());
+
+        if (response.statusCode() < 200 || response.statusCode() >= 300) {
+            throw new IOException(
+                String.format("HTTP call to %s failed with status code %s", uri, response.statusCode()));
+        }
+
+    }
+
+    private static Object toJsonSerializable(Schema<?> schema, Object val) {
+        if (schema == null || schema.getSchemaInfo().getType().isPrimitive()) {
+            return val;
+        }
+        switch (schema.getSchemaInfo().getType()) {
+            case KEY_VALUE:
+                KeyValueSchema<?, ?> keyValueSchema = (KeyValueSchema<?, ?>) schema;
+                org.apache.pulsar.common.schema.KeyValue<?, ?> keyValue =
+                    (org.apache.pulsar.common.schema.KeyValue<?, ?>) val;
+                Map<String, Object> jsonKeyValue = new HashMap<>();
+                Object key = keyValue.getKey();
+                Object value = keyValue.getValue();
+                jsonKeyValue.put("key", toJsonSerializable(keyValueSchema.getKeySchema(),
+                    key instanceof GenericObject ? ((GenericObject) key).getNativeObject() : key));
+                jsonKeyValue.put("value", toJsonSerializable(keyValueSchema.getValueSchema(),
+                    value instanceof GenericObject ? ((GenericObject) value).getNativeObject() : value));
+                return jsonKeyValue;
+            case AVRO:
+                return JsonConverter.toJson((org.apache.avro.generic.GenericRecord) val);
+            case JSON:
+                return val;
+            default:
+                throw new UnsupportedOperationException("Unsupported schema type ="
+                    + schema.getSchemaInfo().getType());
+        }
+    }
+
+    @Override
+    public void close() {}
+}

pulsar-io/http/src/main/java/org/apache/pulsar/io/http/HttpSinkConfig.java

@@ -0,0 +1,58 @@
+/*
+ * 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.pulsar.io.http;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import java.io.File;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+import lombok.Data;
+import lombok.experimental.Accessors;
+import org.apache.pulsar.io.core.annotations.FieldDoc;
+
+@Data
+@Accessors(chain = true)
+public class HttpSinkConfig implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    @FieldDoc(
+        defaultValue = "http://localhost",
+        help = "The URL of the HTTP server")
+    private String url = "http://localhost";
+
+    @FieldDoc(
+        defaultValue = "",
+        help = "The list of default headers added to each request")
+    private Map<String, String> headers = new HashMap<>();
+
+    public static HttpSinkConfig load(String yamlFile) throws IOException {
+        ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
+        return mapper.readValue(new File(yamlFile), HttpSinkConfig.class);
+    }
+
+    public static HttpSinkConfig load(Map<String, Object> map) throws IOException {
+        ObjectMapper mapper = new ObjectMapper();
+        return mapper.readValue(new ObjectMapper().writeValueAsString(map), HttpSinkConfig.class);
+    }
+}