Add a simple pulsar source connector.

flink-connectors/flink-connector-pulsar/pom.xml

@@ -0,0 +1,102 @@
+<?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/maven-v4_0_0.xsd">
+
+	<modelVersion>4.0.0</modelVersion>
+
+	<parent>
+		<groupId>org.apache.flink</groupId>
+		<artifactId>flink-connectors</artifactId>
+		<version>1.6-SNAPSHOT</version>
+		<relativePath>..</relativePath>
+	</parent>
+
+	<artifactId>flink-connector-pulsar_${scala.binary.version}</artifactId>
+	<name>flink-connector-pulsar</name>
+	<properties>
+		<pulsar.version>2.0.0-rc1-incubating</pulsar.version>
+	</properties>
+
+	<packaging>jar</packaging>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+			<version>${project.version}</version>
+			<scope>provided</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-table_${scala.binary.version}</artifactId>
+			<version>${project.version}</version>
+			<scope>provided</scope>
+			<!-- Projects depending on this project, won't depend on flink-table. -->
+			<optional>true</optional>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.pulsar</groupId>
+			<artifactId>pulsar-client</artifactId>
+			<version>${pulsar.version}</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-runtime_${scala.binary.version}</artifactId>
+			<version>${project.version}</version>
+			<type>test-jar</type>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-tests_${scala.binary.version}</artifactId>
+			<version>${project.version}</version>
+			<scope>test</scope>
+			<type>test-jar</type>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+			<version>${project.version}</version>
+			<type>test-jar</type>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-test-utils_${scala.binary.version}</artifactId>
+			<version>${project.version}</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.javassist</groupId>
+			<artifactId>javassist</artifactId>
+			<version>3.20.0-GA</version>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+</project>

flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/streaming/connectors/pulsar/Defaults.java

@@ -0,0 +1,29 @@
+/*
+ * 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.flink.streaming.connectors.pulsar;
+
+/**
+ * Default values for Pulsar connectors.
+ */
+public class Defaults {
+
+	public static final String SERVICE_URL = "pulsar://localhost:6650";
+	public static final long ACKNOWLEDGEMENT_BATCH_SIZE = 100;
+
+	private Defaults() {}
+}

flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/streaming/connectors/pulsar/Metrics.java

@@ -0,0 +1,22 @@
+package org.apache.flink.streaming.connectors.pulsar;
+
+import org.apache.flink.annotation.Internal;
+
+/**
+ * A collection of Pulsar consumer metrics related constant strings.
+ *
+ * <p>The names must not be changed, as that would break backward compatibility for the consumer's metrics.
+ */
+@Internal
+public class Metrics {
+
+	// ------------------------------------------------------------------------
+	//  Per-subtask metrics
+	// ------------------------------------------------------------------------
+
+	public static final String COMMITS_SUCCEEDED_METRICS_COUNTER = "commitsSucceeded";
+	public static final String COMMITS_FAILED_METRICS_COUNTER = "commitsFailed";
+
+	private Metrics() {
+	}
+}

flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/streaming/connectors/pulsar/PulsarConsumerSource.java

@@ -0,0 +1,203 @@
+/*
+ * 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.flink.streaming.connectors.pulsar;
+
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.source.MessageAcknowledgingSourceBase;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+import org.apache.flink.util.IOUtils;
+
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Pulsar source (consumer) which receives messages from a topic and acknowledges messages.
+ * When checkpointing is enabled, it guarantees at least once processing semantics.
+ *
+ * <p>When checkpointing is disabled, it auto acknowledges messages based on the number of messages it has
+ * received. In this mode messages may be dropped.
+ */
+class PulsarConsumerSource<T> extends MessageAcknowledgingSourceBase<T, MessageId> implements PulsarSourceBase<T> {
+
+	private static final Logger LOG = LoggerFactory.getLogger(PulsarConsumerSource.class);
+
+	private final int messageReceiveTimeoutMs = 100;
+	private final String serviceUrl;
+	private final String topic;
+	private final String subscriptionName;
+	private final DeserializationSchema<T> deserializer;
+
+	private PulsarClient client;
+	private Consumer<byte[]> consumer;
+
+	private boolean isCheckpointingEnabled;
+
+	private final long acknowledgementBatchSize;
+	private long batchCount;
+	private long totalMessageCount;
+
+	private transient volatile boolean isRunning;
+
+	PulsarConsumerSource(PulsarSourceBuilder<T> builder) {
+		super(MessageId.class);
+		this.serviceUrl = builder.serviceUrl;
+		this.topic = builder.topic;
+		this.deserializer = builder.deserializationSchema;
+		this.subscriptionName = builder.subscriptionName;
+		this.acknowledgementBatchSize = builder.acknowledgementBatchSize;
+	}
+
+	@Override
+	public void open(Configuration parameters) throws Exception {
+		super.open(parameters);
+
+		final RuntimeContext context = getRuntimeContext();
+		if (context instanceof StreamingRuntimeContext) {
+			isCheckpointingEnabled = ((StreamingRuntimeContext) context).isCheckpointingEnabled();
+		}
+
+		client = createClient();
+		consumer = createConsumer(client);
+
+		isRunning = true;
+	}
+
+	@Override
+	protected void acknowledgeIDs(long checkpointId, Set<MessageId> messageIds) {
+		if (consumer == null) {
+			LOG.error("null consumer unable to acknowledge messages");
+			throw new RuntimeException("null pulsar consumer unable to acknowledge messages");
+		}
+
+		if (messageIds.isEmpty()) {
+			LOG.info("no message ids to acknowledge");
+			return;
+		}
+
+		Map<String, CompletableFuture<Void>> futures = new HashMap<>(messageIds.size());
+		for (MessageId id : messageIds) {
+			futures.put(id.toString(), consumer.acknowledgeAsync(id));
+		}
+
+		futures.forEach((k, f) -> {
+			try {
+				f.get();
+			} catch (Exception e) {
+				LOG.error("failed to acknowledge messageId " + k, e);
+				throw new RuntimeException("Messages could not be acknowledged during checkpoint creation.", e);
+			}
+		});
+	}
+
+	@Override
+	public void run(SourceContext<T> context) throws Exception {
+		Message message;
+		while (isRunning) {
+			message = consumer.receive(messageReceiveTimeoutMs, TimeUnit.MILLISECONDS);
+			if (message == null) {
+				LOG.info("unexpected null message");
+				continue;
+			}
+
+			if (isCheckpointingEnabled) {
+				emitCheckpointing(context, message);
+			} else {
+				emitAutoAcking(context, message);
+			}
+		}
+	}
+
+	private void emitCheckpointing(SourceContext<T> context, Message message) throws IOException {
+		synchronized (context.getCheckpointLock()) {
+			if (!addId(message.getMessageId())) {
+				if (LOG.isDebugEnabled()) {
+					LOG.debug("messageId=" + message.getMessageId().toString() + " already processed.");
+				}
+				return;
+			}
+			context.collect(deserialize(message));
+			totalMessageCount++;
+		}
+	}
+
+	private void emitAutoAcking(SourceContext<T> context, Message message) throws IOException {
+		context.collect(deserialize(message));
+		batchCount++;
+		totalMessageCount++;
+		if (batchCount >= acknowledgementBatchSize) {
+			LOG.info("processed {} messages acknowledging messageId {}", batchCount, message.getMessageId());
+			consumer.acknowledgeCumulative(message.getMessageId());
+			batchCount = 0;
+		}
+	}
+
+	private T deserialize(Message message) throws IOException {
+		return deserializer.deserialize(message.getData());
+	}
+
+	@Override
+	public void cancel() {
+		isRunning = false;
+	}
+
+	@Override
+	public void close() throws Exception {
+		super.close();
+		IOUtils.cleanup(LOG, consumer);
+		IOUtils.cleanup(LOG, client);
+	}
+
+	@Override
+	public TypeInformation<T> getProducedType() {
+		return deserializer.getProducedType();
+	}
+
+	boolean isCheckpointingEnabled() {
+		return isCheckpointingEnabled;
+	}
+
+	PulsarClient createClient() throws PulsarClientException {
+		return PulsarClient.builder()
+			.serviceUrl(serviceUrl)
+			.build();
+	}
+
+	Consumer<byte[]> createConsumer(PulsarClient client) throws PulsarClientException {
+		return client.newConsumer()
+			.topic(topic)
+			.subscriptionName(subscriptionName)
+			.subscriptionType(SubscriptionType.Failover)
+			.subscribe();
+	}
+}

flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/streaming/connectors/pulsar/PulsarSourceBase.java

@@ -0,0 +1,30 @@
+/*
+ * 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.flink.streaming.connectors.pulsar;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
+
+/**
+ * Base class for pulsar sources.
+ * @param <T>
+ */
+@PublicEvolving
+interface PulsarSourceBase<T> extends ParallelSourceFunction<T>, ResultTypeQueryable<T> {
+}