[pulsar-kafka] Support encryption for pulsar-kafka producer/consumer (#26)
* [pulsar-kafka] Support encryption for pulsar-kafka producer/consumer
* add properties param for getEncryptionKey method
diff --git a/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/kafka/clients/producer/PulsarKafkaProducer.java b/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/kafka/clients/producer/PulsarKafkaProducer.java
index f6da38e..fca5da8 100644
--- a/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/kafka/clients/producer/PulsarKafkaProducer.java
+++ b/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/kafka/clients/producer/PulsarKafkaProducer.java
@@ -65,7 +65,7 @@
public class PulsarKafkaProducer<K, V> implements Producer<K, V> {
private final PulsarClient client;
- private final ProducerBuilder<byte[]> pulsarProducerBuilder;
+ final ProducerBuilder<byte[]> pulsarProducerBuilder;
private final ConcurrentMap<String, org.apache.pulsar.client.api.Producer<byte[]>> producers = new ConcurrentHashMap<>();
diff --git a/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/pulsar/client/kafka/compat/CryptoKeyReaderFactory.java b/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/pulsar/client/kafka/compat/CryptoKeyReaderFactory.java
new file mode 100644
index 0000000..649c297
--- /dev/null
+++ b/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/pulsar/client/kafka/compat/CryptoKeyReaderFactory.java
@@ -0,0 +1,51 @@
+/**
+ * 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.client.kafka.compat;
+
+import java.util.Properties;
+import java.util.Set;
+
+import org.apache.pulsar.client.api.CryptoKeyReader;
+
+/**
+ * Factory class to create {@link CryptoKeyReader} by using configuration stored in Producer/Consumer properties.
+ *
+ */
+public interface CryptoKeyReaderFactory {
+
+ /**
+ * Create CryptoKeyReader object for Producer/Consumer.
+ *
+ * @param properties
+ * properties provided by user to create CryptoKeyReader based on configuration params
+ * @return CryptoKeyReader
+ */
+ CryptoKeyReader create(Properties properties);
+
+ /**
+ * Encryption keys for {@link CryptoKeyReader} while enabling encryption at producer.
+ *
+ * @param properties
+ * properties provided by user to get encryption-keys based on configuration params
+ * @return Set of encryption keys
+ */
+ default Set<String> getEncryptionKey(Properties properties) {
+ return null;
+ }
+}
diff --git a/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/pulsar/client/kafka/compat/PulsarConsumerKafkaConfig.java b/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/pulsar/client/kafka/compat/PulsarConsumerKafkaConfig.java
index 09a9806..40858b1 100644
--- a/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/pulsar/client/kafka/compat/PulsarConsumerKafkaConfig.java
+++ b/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/pulsar/client/kafka/compat/PulsarConsumerKafkaConfig.java
@@ -22,6 +22,7 @@
import java.util.Properties;
import java.util.concurrent.TimeUnit;
import static org.apache.pulsar.client.kafka.compat.PulsarProducerKafkaConfig.AUTO_UPDATE_PARTITIONS;
+import static org.apache.pulsar.client.kafka.compat.PulsarProducerKafkaConfig.CRYPTO_READER_FACTORY_CLASS_NAME;
import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.RegexSubscriptionMode;
@@ -70,6 +71,17 @@
if (properties.containsKey(AUTO_UPDATE_PARTITIONS)) {
consumerBuilder.autoUpdatePartitions(Boolean.parseBoolean(properties.getProperty(AUTO_UPDATE_PARTITIONS)));
}
+
+ if (properties.containsKey(CRYPTO_READER_FACTORY_CLASS_NAME)) {
+ try {
+ CryptoKeyReaderFactory cryptoReaderFactory = (CryptoKeyReaderFactory) Class
+ .forName(properties.getProperty(CRYPTO_READER_FACTORY_CLASS_NAME)).newInstance();
+ consumerBuilder.cryptoKeyReader(cryptoReaderFactory.create(properties));
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Failed to create crypto reader using factory "
+ + properties.getProperty(CRYPTO_READER_FACTORY_CLASS_NAME), e);
+ }
+ }
return consumerBuilder;
}
}
diff --git a/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/pulsar/client/kafka/compat/PulsarProducerKafkaConfig.java b/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/pulsar/client/kafka/compat/PulsarProducerKafkaConfig.java
index 3315cd2..276d777 100644
--- a/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/pulsar/client/kafka/compat/PulsarProducerKafkaConfig.java
+++ b/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/pulsar/client/kafka/compat/PulsarProducerKafkaConfig.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.client.kafka.compat;
import java.util.Properties;
+import java.util.Set;
import org.apache.pulsar.client.api.ProducerBuilder;
import org.apache.pulsar.client.api.PulsarClient;
@@ -35,6 +36,7 @@
public static final String BATCHING_ENABLED = "pulsar.producer.batching.enabled";
public static final String BATCHING_MAX_MESSAGES = "pulsar.producer.batching.max.messages";
public static final String AUTO_UPDATE_PARTITIONS = "pulsar.auto.update.partitions";
+ public static final String CRYPTO_READER_FACTORY_CLASS_NAME = "pulsar.crypto.reader.factory.class.name";
/**
* send operations will immediately fail with {@link ProducerQueueIsFullError} when there is no space left in
* pending queue.
@@ -70,6 +72,21 @@
if (properties.containsKey(AUTO_UPDATE_PARTITIONS)) {
producerBuilder.autoUpdatePartitions(Boolean.parseBoolean(properties.getProperty(AUTO_UPDATE_PARTITIONS)));
}
+
+ if (properties.containsKey(CRYPTO_READER_FACTORY_CLASS_NAME)) {
+ try {
+ CryptoKeyReaderFactory cryptoReaderFactory = (CryptoKeyReaderFactory) Class
+ .forName(properties.getProperty(CRYPTO_READER_FACTORY_CLASS_NAME)).newInstance();
+ producerBuilder.cryptoKeyReader(cryptoReaderFactory.create(properties));
+ Set<String> keys = cryptoReaderFactory.getEncryptionKey(properties);
+ if (keys != null) {
+ keys.forEach(producerBuilder::addEncryptionKey);
+ }
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Failed to create crypto reader using factory "
+ + properties.getProperty(CRYPTO_READER_FACTORY_CLASS_NAME), e);
+ }
+ }
return producerBuilder;
}
}
diff --git a/pulsar-client-kafka-compat/pulsar-client-kafka/src/test/java/org/apache/kafka/clients/producer/PulsarCliebtKafkaConfigTest.java b/pulsar-client-kafka-compat/pulsar-client-kafka/src/test/java/org/apache/kafka/clients/producer/PulsarCliebtKafkaConfigTest.java
new file mode 100644
index 0000000..5ddddf7
--- /dev/null
+++ b/pulsar-client-kafka-compat/pulsar-client-kafka/src/test/java/org/apache/kafka/clients/producer/PulsarCliebtKafkaConfigTest.java
@@ -0,0 +1,111 @@
+/**
+ * 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.kafka.clients.producer;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.testng.Assert.assertEquals;
+
+import java.util.Arrays;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.CryptoKeyReader;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.impl.ConsumerBuilderImpl;
+import org.apache.pulsar.client.impl.DefaultCryptoKeyReader;
+import org.apache.pulsar.client.impl.ProducerBuilderImpl;
+import org.apache.pulsar.client.kafka.compat.CryptoKeyReaderFactory;
+import org.apache.pulsar.client.kafka.compat.PulsarConsumerKafkaConfig;
+import org.apache.pulsar.client.kafka.compat.PulsarProducerKafkaConfig;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.google.common.collect.Sets;
+
+public class PulsarCliebtKafkaConfigTest {
+
+
+ @Test
+ public void testPulsarKafkaProducer() {
+ ClientBuilder mockClientBuilder = mock(ClientBuilder.class);
+ ProducerBuilder mockProducerBuilder = mock(ProducerBuilder.class);
+ doAnswer(invocation -> {
+ Assert.assertEquals((int)invocation.getArguments()[0], 1000000, "Send time out is suppose to be 1000.");
+ return mockProducerBuilder;
+ }).when(mockProducerBuilder).sendTimeout(anyInt(), any(TimeUnit.class));
+ doReturn(mockClientBuilder).when(mockClientBuilder).serviceUrl(anyString());
+ doAnswer(invocation -> {
+ Assert.assertEquals((int)invocation.getArguments()[0], 1000, "Keep alive interval is suppose to be 1000.");
+ return mockClientBuilder;
+ }).when(mockClientBuilder).keepAliveInterval(anyInt(), any(TimeUnit.class));
+
+ // validate producer
+ Properties properties = new Properties();
+ properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, DefaultPartitioner.class);
+ properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Arrays.asList("pulsar://localhost:6650"));
+ properties.put(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG, "1000000");
+ properties.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, "1000000");
+ properties.put(PulsarProducerKafkaConfig.CRYPTO_READER_FACTORY_CLASS_NAME, CryptoKeyReaderFactoryImpl.class.getName());
+ PulsarKafkaProducer producer = new PulsarKafkaProducer<>(properties);
+ ProducerBuilderImpl producerBuilder = (ProducerBuilderImpl) producer.pulsarProducerBuilder;
+ assertEquals(producerBuilder.getConf().getCryptoKeyReader(), CryptoKeyReaderFactoryImpl.reader);
+ assertEquals(producerBuilder.getConf().getEncryptionKeys(), CryptoKeyReaderFactoryImpl.encryptionKeys);
+
+ // validate consumer
+ properties = new Properties();
+ properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Arrays.asList("pulsar://localhost:6650"));
+ properties.put(PulsarProducerKafkaConfig.CRYPTO_READER_FACTORY_CLASS_NAME, CryptoKeyReaderFactoryImpl.class.getName());
+ PulsarClient client = mock(PulsarClient.class);
+ ConsumerBuilderImpl<byte[]> consumerBuilder = new ConsumerBuilderImpl<>(null, null);
+ doReturn(consumerBuilder).when(client).newConsumer();
+ PulsarConsumerKafkaConfig.getConsumerBuilder(client , properties);
+ assertEquals(consumerBuilder.getConf().getCryptoKeyReader(), CryptoKeyReaderFactoryImpl.reader);
+ }
+
+ public static class CryptoKeyReaderFactoryImpl implements CryptoKeyReaderFactory {
+ static final CryptoKeyReader reader = DefaultCryptoKeyReader.builder().build();
+ static final Set<String> encryptionKeys = Sets.newHashSet("test");
+
+ @Override
+ public CryptoKeyReader create(Properties properties) {
+ return reader;
+ }
+
+ @Override
+ public Set<String> getEncryptionKey(Properties properties) {
+ return encryptionKeys;
+ }
+ }
+}
diff --git a/pulsar-client-kafka-compat/pulsar-client-kafka/src/test/java/org/apache/kafka/clients/producer/PulsarKafkaProducerTest.java b/pulsar-client-kafka-compat/pulsar-client-kafka/src/test/java/org/apache/kafka/clients/producer/PulsarKafkaProducerTest.java
index b1069ad..bf6a6cf 100644
--- a/pulsar-client-kafka-compat/pulsar-client-kafka/src/test/java/org/apache/kafka/clients/producer/PulsarKafkaProducerTest.java
+++ b/pulsar-client-kafka-compat/pulsar-client-kafka/src/test/java/org/apache/kafka/clients/producer/PulsarKafkaProducerTest.java
@@ -270,5 +270,4 @@
}
}
-
}
