[pulsar-io] Add NSQ Source (apache#8250)

### Motivation

This PR adds a source representing an [NSQ](https://nsq.io/) topic to be mirrored to a pulsar topic.

### Modifications

With the exception of adding it as a module to `pulsar-io/pom.xml`, there are no modifications to existing code.

distribution/io/src/assemble/io.xml

@@ -50,6 +50,7 @@
     <file><source>${basedir}/../../pulsar-io/kafka/target/pulsar-io-kafka-${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>
     <file><source>${basedir}/../../pulsar-io/jdbc/sqlite/target/pulsar-io-jdbc-sqlite-${project.version}.nar</source></file>
     <file><source>${basedir}/../../pulsar-io/jdbc/mariadb/target/pulsar-io-jdbc-mariadb-${project.version}.nar</source></file>
     <file><source>${basedir}/../../pulsar-io/jdbc/clickhouse/target/pulsar-io-jdbc-clickhouse-${project.version}.nar</source></file>

pom.xml

@@ -180,6 +180,7 @@ flexible messaging model and an intuitive client API.</description>
     <javax.activation.version>1.2.0</javax.activation.version>
     <jna.version>4.2.0</jna.version>
     <kubernetesclient.version>9.0.2</kubernetesclient.version>
+    <nsq-client.version>1.0</nsq-client.version>
 
     <!-- test dependencies -->
     <cassandra.version>3.6.0</cassandra.version>

pulsar-io/nsq/pom.xml

@@ -0,0 +1,74 @@
+<!--
+
+    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.7.0-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>pulsar-io-nsq</artifactId>
+  <name>Pulsar IO :: NSQ</name>
+
+  <dependencies>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-io-core</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>com.sproutsocial</groupId>
+      <artifactId>nsq-j</artifactId>
+      <version>${nsq-client.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>commons-collections</groupId>
+      <artifactId>commons-collections</artifactId>
+      <version>${commons.collections.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-lang3</artifactId>
+      <version>3.4</version>
+    </dependency>
+
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-io-common</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
+  </dependencies>
+
+  <build>
+      <plugins>
+        <plugin>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-nar-maven-plugin</artifactId>
+        </plugin>
+      </plugins>
+  </build>
+</project>

pulsar-io/nsq/src/main/java/org/apache/pulsar/io/nsq/NSQSource.java

@@ -0,0 +1,109 @@
+/**
+ * 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.nsq;
+
+import java.io.IOException;
+import java.util.Map;
+
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.pulsar.io.common.IOConfigUtils;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.io.core.PushSource;
+import org.apache.pulsar.io.core.SourceContext;
+import org.apache.pulsar.io.core.annotations.Connector;
+import org.apache.pulsar.io.core.annotations.IOType;
+
+import com.sproutsocial.nsq.Client;
+import com.sproutsocial.nsq.Subscriber;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Connector(
+    name = "nsq",
+    type = IOType.SOURCE,
+    help = "A Simple connector moving messages from an NSQ topic to a Pulsar Topic",
+    configClass = NSQSourceConfig.class
+)
+@Slf4j
+public class NSQSource extends PushSource<byte[]> {
+
+    private static final Logger LOG = LoggerFactory.getLogger(NSQSource.class);
+
+    private Subscriber subscriber;
+
+    private Object waitObject;
+
+    @Override
+    public void open(Map<String, Object> config, SourceContext sourceContext) throws IOException {
+        NSQSourceConfig nsqSourceConfig = IOConfigUtils.loadWithSecrets(config, NSQSourceConfig.class, sourceContext);
+        nsqSourceConfig.validate();
+
+        waitObject = new Object();
+        startThread(nsqSourceConfig);
+    }
+
+    @Override
+    public void close() throws Exception{
+        stopThread();
+    }
+
+    private void startThread(NSQSourceConfig config) {
+        String[] lookupds = new String[config.getLookupds().size()];
+        config.getLookupds().toArray(lookupds);
+        subscriber = new Subscriber(lookupds);
+
+        Thread runnerThread = new Thread(() -> {
+            subscriber.subscribe(config.getTopic(), config.getChannel(), (byte[]data) ->{
+                consume(new NSQRecord(data));
+            });
+            LOG.info("NSQ Consumer started for topic {} with channel {}", config.getTopic(), config.getChannel());
+            //wait
+            try {
+                synchronized (waitObject) {
+                    waitObject.wait();
+                }
+            } catch (Exception e) {
+                LOG.info("Got an exception in waitObject");
+            }
+            LOG.debug("Closing the NSQ connection");
+            subscriber.stop();
+            Client.getDefaultClient().stop();
+            LOG.info("NSQ subscriber stopped");
+            LOG.info("NSQ Runner Thread ending");
+        });
+        runnerThread.setName("NSQSubscriberRunner");
+        runnerThread.start();
+    }
+
+    private void stopThread() {
+        LOG.info("Source closed");
+        synchronized (waitObject) {
+            waitObject.notify();
+        }
+    }
+
+    @Data
+    static private class NSQRecord implements Record<byte[]> {
+        private final byte[] value;
+    }
+}
+

pulsar-io/nsq/src/main/java/org/apache/pulsar/io/nsq/NSQSourceConfig.java

@@ -0,0 +1,112 @@
+/**
+ * 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.nsq;
+
+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.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+
+import lombok.Data;
+import lombok.experimental.Accessors;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.io.core.annotations.FieldDoc;
+
+/**
+ * Configuration object for the NSQ Connector.
+ */
+@Data
+@Accessors(chain=true)
+public class NSQSourceConfig implements Serializable {
+    private static final long serialVersionUID = 1L;
+
+
+    @FieldDoc(
+        required= true,
+        defaultValue = "",
+        help = "The topic you wish to transport into pulsar"
+    )
+    private String topic;
+
+    @FieldDoc(
+        required= false,
+        defaultValue = "pulsar-transport-<topic>",
+        help = "The channel to use on the topic you want to transport"
+    )
+    private String channel;
+
+    @FieldDoc(
+        required= true,
+        defaultValue = "",
+        help = "A comma-separated list of nsqlookupd hosts to contact"
+    )
+    private String lookupds;
+
+    public static NSQSourceConfig load(String yamlFile) throws IOException {
+        ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
+        return applyDefaults(mapper.readValue(new File(yamlFile), NSQSourceConfig.class));
+    }
+
+    public static NSQSourceConfig load(Map<String, Object> map) throws IOException {
+        ObjectMapper mapper = new ObjectMapper();
+        return applyDefaults(mapper.readValue(new ObjectMapper().writeValueAsString(map), NSQSourceConfig.class));
+    }
+
+    private static NSQSourceConfig applyDefaults(NSQSourceConfig config) {
+        if (config.channel == null) {
+            config.channel=String.format("pulsar-transport-%s", config.topic);
+        }
+        return config;
+    }
+
+
+    public void validate() throws IllegalArgumentException {
+        if (getChannel() == null) {
+            setChannel(String.format("pulsar-transport-%s", getTopic()));
+        }
+        if (getTopic() == null || getLookupds() == null || getChannel() == null){
+            throw new IllegalArgumentException("Required property not set.");
+        }
+    }
+
+    public List<String> getLookupds(){
+        if (StringUtils.isBlank(lookupds)){
+            return Collections.emptyList();
+        }
+
+        List<String> out = new ArrayList<String> ();
+        for (String s: StringUtils.split(lookupds, ",")) {
+            out.add(StringUtils.trim(s));
+        }
+
+        if (CollectionUtils.isEmpty(out)){
+            return Collections.emptyList();
+        }
+        return out;
+    }
+}

pulsar-io/nsq/src/main/resources/META-INF/services/pulsar-io.yml

@@ -0,0 +1,4 @@
+name: nsq
+description: Ingest data from an NSQ topic
+sourceClass: org.apache.pulsar.io.nsq.NSQSource
+sourceConfigClass: org.apache.pulsar.io.nsq.NSQConfig