You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zh...@apache.org on 2019/02/11 09:57:33 UTC
[pulsar] branch master updated: Add a unit test for
KafkaAbstractSink. (#3564)
This is an automated email from the ASF dual-hosted git repository.
zhaijia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 986ea8b Add a unit test for KafkaAbstractSink. (#3564)
986ea8b is described below
commit 986ea8b6eaa7b16bda5d5f89db25a2aa5bf47a12
Author: Fangbin Sun <su...@gmail.com>
AuthorDate: Mon Feb 11 17:57:28 2019 +0800
Add a unit test for KafkaAbstractSink. (#3564)
### Motivation
Add a unit test for KafkaAbstractSink to check valid configs.
### Modifications
Add KafkaAbstractSinkTest.
---
.../io/kafka/sink/KafkaAbstractSinkTest.java | 168 +++++++++++++++++++++
.../kafka/src/test/resources/kafkaSinkConfig.yaml | 30 ++++
2 files changed, 198 insertions(+)
diff --git a/pulsar-io/kafka/src/test/java/org/apache/pulsar/io/kafka/sink/KafkaAbstractSinkTest.java b/pulsar-io/kafka/src/test/java/org/apache/pulsar/io/kafka/sink/KafkaAbstractSinkTest.java
new file mode 100644
index 0000000..2295820
--- /dev/null
+++ b/pulsar-io/kafka/src/test/java/org/apache/pulsar/io/kafka/sink/KafkaAbstractSinkTest.java
@@ -0,0 +1,168 @@
+/**
+ * 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.kafka.sink;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.io.core.KeyValue;
+import org.apache.pulsar.io.core.SinkContext;
+import org.apache.pulsar.io.kafka.KafkaAbstractSink;
+import org.apache.pulsar.io.kafka.KafkaSinkConfig;
+import org.slf4j.Logger;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.testng.Assert.*;
+
+public class KafkaAbstractSinkTest {
+ private static class DummySink extends KafkaAbstractSink<String, byte[]> {
+
+ @Override
+ public KeyValue extractKeyValue(Record record) {
+ return new KeyValue<>(record.getKey().orElse(null), record.getValue());
+ }
+ }
+
+ @FunctionalInterface
+ public interface ThrowingRunnable {
+ void run() throws Throwable;
+ }
+
+ private static <T extends Exception> void expectThrows(Class<T> expectedType, String expectedMessage, ThrowingRunnable runnable) {
+ try {
+ runnable.run();
+ Assert.fail();
+ } catch (Throwable e) {
+ if (expectedType.isInstance(e)) {
+ T ex = expectedType.cast(e);
+ assertEquals(expectedMessage, ex.getMessage());
+ return;
+ }
+ throw new AssertionError("Unexpected exception type, expected " + expectedType.getSimpleName() + " but got " + e);
+ }
+ throw new AssertionError("Expected exception");
+ }
+
+ @Test
+ public void testInvalidConfigWillThrownException() throws Exception {
+ KafkaAbstractSink sink = new DummySink();
+ Map<String, Object> config = new HashMap<>();
+ SinkContext sc = new SinkContext() {
+ @Override
+ public int getInstanceId() {
+ return 0;
+ }
+
+ @Override
+ public int getNumInstances() {
+ return 0;
+ }
+
+ @Override
+ public void recordMetric(String metricName, double value) {
+
+ }
+
+ @Override
+ public Collection<String> getInputTopics() {
+ return null;
+ }
+
+ @Override
+ public String getTenant() {
+ return null;
+ }
+
+ @Override
+ public String getNamespace() {
+ return null;
+ }
+
+ @Override
+ public String getSinkName() {
+ return null;
+ }
+
+ @Override
+ public Logger getLogger() {
+ return null;
+ }
+ };
+ ThrowingRunnable openAndClose = ()->{
+ try {
+ sink.open(config, sc);
+ fail();
+ } finally {
+ sink.close();
+ }
+ };
+ expectThrows(NullPointerException.class, "Kafka topic is not set", openAndClose);
+ config.put("topic", "topic_2");
+ expectThrows(NullPointerException.class, "Kafka bootstrapServers is not set", openAndClose);
+ config.put("bootstrapServers", "localhost:6667");
+ expectThrows(NullPointerException.class, "Kafka acks mode is not set", openAndClose);
+ config.put("acks", "1");
+ config.put("batchSize", "-1");
+ expectThrows(IllegalArgumentException.class, "Invalid Kafka Producer batchSize : -1", openAndClose);
+ config.put("batchSize", "16384");
+ config.put("maxRequestSize", "-1");
+ expectThrows(IllegalArgumentException.class, "Invalid Kafka Producer maxRequestSize : -1", openAndClose);
+ config.put("maxRequestSize", "1048576");
+ config.put("acks", "none");
+ expectThrows(ConfigException.class, "Invalid value none for configuration acks: String must be one of: all, -1, 0, 1", openAndClose);
+ config.put("acks", "1");
+ sink.open(config, sc);
+ sink.close();
+ }
+
+ @Test
+ public final void loadFromYamlFileTest() throws IOException {
+ File yamlFile = getFile("kafkaSinkConfig.yaml");
+ KafkaSinkConfig config = KafkaSinkConfig.load(yamlFile.getAbsolutePath());
+ assertNotNull(config);
+ assertEquals("localhost:6667", config.getBootstrapServers());
+ assertEquals("test", config.getTopic());
+ assertEquals("1", config.getAcks());
+ assertEquals(Long.parseLong("16384"), config.getBatchSize());
+ assertEquals(Long.parseLong("1048576"), config.getMaxRequestSize());
+ assertNotNull(config.getProducerConfigProperties());
+ Properties props = new Properties();
+ props.putAll(config.getProducerConfigProperties());
+ props.put(ProducerConfig.ACKS_CONFIG, config.getAcks());
+ assertEquals("test-pulsar-producer", props.getProperty("client.id"));
+ assertEquals("SASL_PLAINTEXT", props.getProperty("security.protocol"));
+ assertEquals("GSSAPI", props.getProperty("sasl.mechanism"));
+ assertEquals("1", props.getProperty(ProducerConfig.ACKS_CONFIG));
+ }
+
+ private File getFile(String name) {
+ ClassLoader classLoader = getClass().getClassLoader();
+ return new File(classLoader.getResource(name).getFile());
+ }
+
+}
diff --git a/pulsar-io/kafka/src/test/resources/kafkaSinkConfig.yaml b/pulsar-io/kafka/src/test/resources/kafkaSinkConfig.yaml
new file mode 100644
index 0000000..8f073f8
--- /dev/null
+++ b/pulsar-io/kafka/src/test/resources/kafkaSinkConfig.yaml
@@ -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.
+#
+
+"bootstrapServers": "localhost:6667"
+"topic": "test"
+"acks": "1"
+"batchSize": "16384"
+"maxRequestSize": "1048576"
+"producerConfigProperties":
+ "client.id": "test-pulsar-producer"
+ "security.protocol": "SASL_PLAINTEXT"
+ "sasl.mechanism": "GSSAPI"
+ "sasl.kerberos.service.name": "kafka"
+ "acks": "all"
\ No newline at end of file