| /** |
| * 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.tests.integration.compat.kafka; |
| |
| import static org.testng.Assert.assertEquals; |
| import static org.testng.Assert.assertNull; |
| |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Properties; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicInteger; |
| |
| import lombok.Cleanup; |
| import lombok.Data; |
| import lombok.EqualsAndHashCode; |
| import lombok.ToString; |
| import lombok.extern.slf4j.Slf4j; |
| |
| import org.apache.avro.reflect.Nullable; |
| import org.apache.kafka.clients.consumer.Consumer; |
| import org.apache.kafka.clients.consumer.ConsumerRecords; |
| import org.apache.kafka.clients.consumer.KafkaConsumer; |
| import org.apache.kafka.clients.consumer.OffsetAndMetadata; |
| import org.apache.kafka.clients.producer.KafkaProducer; |
| import org.apache.kafka.clients.producer.Partitioner; |
| import org.apache.kafka.clients.producer.Producer; |
| import org.apache.kafka.clients.producer.ProducerRecord; |
| import org.apache.kafka.clients.producer.RecordMetadata; |
| import org.apache.kafka.common.Cluster; |
| import org.apache.kafka.common.TopicPartition; |
| import org.apache.kafka.common.serialization.IntegerDeserializer; |
| import org.apache.kafka.common.serialization.IntegerSerializer; |
| import org.apache.kafka.common.serialization.StringDeserializer; |
| import org.apache.kafka.common.serialization.StringSerializer; |
| import org.apache.pulsar.client.admin.PulsarAdmin; |
| import org.apache.pulsar.client.api.Message; |
| import org.apache.pulsar.client.api.PulsarClient; |
| import org.apache.pulsar.client.api.Schema; |
| import org.apache.pulsar.client.api.schema.SchemaDefinition; |
| import org.apache.pulsar.client.impl.schema.AvroSchema; |
| import org.apache.pulsar.client.impl.schema.JSONSchema; |
| import org.apache.pulsar.client.impl.schema.StringSchema; |
| import org.apache.pulsar.client.kafka.compat.PulsarKafkaSchema; |
| import org.apache.pulsar.tests.integration.suites.PulsarStandaloneTestSuite; |
| import org.testng.Assert; |
| import org.testng.annotations.Test; |
| |
| @Slf4j |
| public class KafkaApiTest extends PulsarStandaloneTestSuite { |
| |
| @Data |
| public static class Foo { |
| @Nullable |
| private String field1; |
| @Nullable |
| private String field2; |
| private int field3; |
| } |
| |
| @Data |
| public static class Bar { |
| private boolean field1; |
| } |
| |
| private static String getPlainTextServiceUrl() { |
| return container.getPlainTextServiceUrl(); |
| } |
| |
| private static String getHttpServiceUrl() { |
| return container.getHttpServiceUrl(); |
| } |
| |
| @Test(timeOut = 30000) |
| public void testSimpleProducerConsumer() throws Exception { |
| String topic = "persistent://public/default/testSimpleProducerConsumer"; |
| |
| Properties producerProperties = new Properties(); |
| producerProperties.put("bootstrap.servers", getPlainTextServiceUrl()); |
| producerProperties.put("key.serializer", IntegerSerializer.class.getName()); |
| producerProperties.put("value.serializer", StringSerializer.class.getName()); |
| Producer<Integer, String> producer = new KafkaProducer<>(producerProperties); |
| |
| Properties consumerProperties = new Properties(); |
| consumerProperties.put("bootstrap.servers", getPlainTextServiceUrl()); |
| consumerProperties.put("group.id", "my-subscription-name"); |
| consumerProperties.put("key.deserializer", IntegerDeserializer.class.getName()); |
| consumerProperties.put("value.deserializer", StringDeserializer.class.getName()); |
| consumerProperties.put("enable.auto.commit", "true"); |
| Consumer<Integer, String> consumer = new KafkaConsumer<>(consumerProperties); |
| consumer.subscribe(Arrays.asList(topic)); |
| |
| List<Long> offsets = new ArrayList<>(); |
| |
| for (int i = 0; i < 10; i++) { |
| RecordMetadata md = producer.send(new ProducerRecord<Integer, String>(topic, i, "hello-" + i)).get(); |
| offsets.add(md.offset()); |
| log.info("Published message at {}", Long.toHexString(md.offset())); |
| } |
| |
| producer.flush(); |
| producer.close(); |
| |
| AtomicInteger received = new AtomicInteger(); |
| while (received.get() < 10) { |
| ConsumerRecords<Integer, String> records = consumer.poll(100); |
| records.forEach(record -> { |
| assertEquals(record.key().intValue(), received.get()); |
| assertEquals(record.value(), "hello-" + received.get()); |
| assertEquals(record.offset(), offsets.get(received.get()).longValue()); |
| |
| received.incrementAndGet(); |
| }); |
| |
| consumer.commitSync(); |
| } |
| |
| consumer.close(); |
| } |
| |
| @Test |
| public void testSimpleConsumer() throws Exception { |
| String topic = "testSimpleConsumer"; |
| |
| Properties props = new Properties(); |
| props.put("bootstrap.servers", getPlainTextServiceUrl()); |
| props.put("group.id", "my-subscription-name"); |
| props.put("enable.auto.commit", "false"); |
| props.put("key.deserializer", StringDeserializer.class.getName()); |
| props.put("value.deserializer", StringDeserializer.class.getName()); |
| |
| @Cleanup |
| Consumer<String, String> consumer = new KafkaConsumer<>(props); |
| consumer.subscribe(Arrays.asList(topic)); |
| |
| @Cleanup |
| PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPlainTextServiceUrl()).build(); |
| |
| @Cleanup |
| org.apache.pulsar.client.api.Producer<byte[]> pulsarProducer = pulsarClient.newProducer().topic(topic).create(); |
| |
| for (int i = 0; i < 10; i++) { |
| pulsarProducer.newMessage().key(Integer.toString(i)).value(("hello-" + i).getBytes()).send(); |
| } |
| |
| AtomicInteger received = new AtomicInteger(); |
| while (received.get() < 10) { |
| ConsumerRecords<String, String> records = consumer.poll(100); |
| if (!records.isEmpty()) { |
| records.forEach(record -> { |
| String key = Integer.toString(received.get()); |
| String value = "hello-" + received.get(); |
| log.info("Receive record : key = {}, value = {}, topic = {}, ptn = {}", |
| key, value, record.topic(), record.partition()); |
| assertEquals(record.key(), key); |
| assertEquals(record.value(), value); |
| |
| received.incrementAndGet(); |
| }); |
| |
| consumer.commitSync(); |
| } |
| } |
| } |
| |
| @Test |
| public void testConsumerAutoCommit() throws Exception { |
| String topic = "testConsumerAutoCommit"; |
| |
| Properties props = new Properties(); |
| props.put("bootstrap.servers", getPlainTextServiceUrl()); |
| props.put("group.id", "my-subscription-name"); |
| props.put("enable.auto.commit", "true"); |
| props.put("key.deserializer", StringDeserializer.class.getName()); |
| props.put("value.deserializer", StringDeserializer.class.getName()); |
| |
| Consumer<String, String> consumer = new KafkaConsumer<>(props); |
| consumer.subscribe(Arrays.asList(topic)); |
| |
| @Cleanup |
| PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPlainTextServiceUrl()).build(); |
| org.apache.pulsar.client.api.Producer<byte[]> pulsarProducer = pulsarClient.newProducer().topic(topic).create(); |
| |
| for (int i = 0; i < 10; i++) { |
| pulsarProducer.newMessage().key(Integer.toString(i)).value(("hello-" + i).getBytes()).send(); |
| } |
| |
| AtomicInteger received = new AtomicInteger(); |
| while (received.get() < 10) { |
| ConsumerRecords<String, String> records = consumer.poll(100); |
| records.forEach(record -> { |
| assertEquals(record.key(), Integer.toString(received.get())); |
| assertEquals(record.value(), "hello-" + received.get()); |
| received.incrementAndGet(); |
| }); |
| } |
| |
| consumer.close(); |
| |
| // Re-open consumer and verify every message was acknowledged |
| Consumer<String, String> consumer2 = new KafkaConsumer<>(props); |
| consumer2.subscribe(Arrays.asList(topic)); |
| |
| ConsumerRecords<String, String> records = consumer2.poll(100); |
| assertEquals(records.count(), 0); |
| consumer2.close(); |
| } |
| |
| @Test |
| public void testConsumerManualOffsetCommit() throws Exception { |
| String topic = "testConsumerManualOffsetCommit"; |
| |
| Properties props = new Properties(); |
| props.put("bootstrap.servers", getPlainTextServiceUrl()); |
| props.put("group.id", "my-subscription-name"); |
| props.put("enable.auto.commit", "false"); |
| props.put("key.deserializer", StringDeserializer.class.getName()); |
| props.put("value.deserializer", StringDeserializer.class.getName()); |
| |
| Consumer<String, String> consumer = new KafkaConsumer<>(props); |
| consumer.subscribe(Arrays.asList(topic)); |
| |
| @Cleanup |
| PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPlainTextServiceUrl()).build(); |
| org.apache.pulsar.client.api.Producer<byte[]> pulsarProducer = pulsarClient.newProducer().topic(topic).create(); |
| |
| for (int i = 0; i < 10; i++) { |
| pulsarProducer.newMessage().key(Integer.toString(i)).value(("hello-" + i).getBytes()).send(); |
| } |
| |
| AtomicInteger received = new AtomicInteger(); |
| while (received.get() < 10) { |
| ConsumerRecords<String, String> records = consumer.poll(100); |
| records.forEach(record -> { |
| assertEquals(record.key(), Integer.toString(received.get())); |
| assertEquals(record.value(), "hello-" + received.get()); |
| |
| Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(); |
| offsets.put(new TopicPartition(record.topic(), record.partition()), |
| new OffsetAndMetadata(record.offset())); |
| consumer.commitSync(offsets); |
| |
| received.incrementAndGet(); |
| }); |
| } |
| |
| consumer.close(); |
| |
| // Re-open consumer and verify every message was acknowledged |
| Consumer<String, String> consumer2 = new KafkaConsumer<>(props); |
| consumer2.subscribe(Arrays.asList(topic)); |
| |
| ConsumerRecords<String, String> records = consumer2.poll(100); |
| assertEquals(records.count(), 0); |
| consumer2.close(); |
| } |
| |
| @Test |
| public void testPartitions() throws Exception { |
| String topic = "testPartitions"; |
| |
| // Create 8 partitions in topic |
| @Cleanup |
| PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(getHttpServiceUrl()).build(); |
| admin.topics().createPartitionedTopic(topic, 8); |
| |
| Properties props = new Properties(); |
| props.put("bootstrap.servers", getPlainTextServiceUrl()); |
| props.put("group.id", "my-subscription-name"); |
| props.put("enable.auto.commit", "true"); |
| props.put("key.deserializer", StringDeserializer.class.getName()); |
| props.put("value.deserializer", StringDeserializer.class.getName()); |
| |
| @Cleanup |
| PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPlainTextServiceUrl()).build(); |
| org.apache.pulsar.client.api.Producer<byte[]> pulsarProducer = pulsarClient.newProducer().topic(topic) |
| .messageRoutingMode(org.apache.pulsar.client.api.MessageRoutingMode.RoundRobinPartition).create(); |
| |
| // Create 2 Kakfa consumer and verify each gets half of the messages |
| List<Consumer<String, String>> consumers = new ArrayList<>(); |
| for (int c = 0; c < 2; c++) { |
| Consumer<String, String> consumer = new KafkaConsumer<>(props); |
| consumer.subscribe(Arrays.asList(topic)); |
| consumers.add(consumer); |
| } |
| |
| int N = 8 * 3; |
| |
| for (int i = 0; i < N; i++) { |
| pulsarProducer.newMessage().key(Integer.toString(i)).value(("hello-" + i).getBytes()).send(); |
| } |
| |
| consumers.forEach(consumer -> { |
| int expectedMessaged = N / consumers.size(); |
| |
| for (int i = 0; i < expectedMessaged;) { |
| ConsumerRecords<String, String> records = consumer.poll(100); |
| i += records.count(); |
| } |
| |
| // No more messages for this consumer |
| ConsumerRecords<String, String> records = consumer.poll(100); |
| assertEquals(records.count(), 0); |
| }); |
| |
| consumers.forEach(Consumer::close); |
| } |
| |
| @Test |
| public void testExplicitPartitions() throws Exception { |
| String topic = "testExplicitPartitions"; |
| |
| // Create 8 partitions in topic |
| @Cleanup |
| PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(getHttpServiceUrl()).build(); |
| admin.topics().createPartitionedTopic(topic, 8); |
| |
| Properties producerProperties = new Properties(); |
| producerProperties.put("bootstrap.servers", getPlainTextServiceUrl()); |
| producerProperties.put("key.serializer", IntegerSerializer.class.getName()); |
| producerProperties.put("value.serializer", StringSerializer.class.getName()); |
| |
| @Cleanup |
| Producer<Integer, String> producer = new KafkaProducer<>(producerProperties); |
| |
| Properties props = new Properties(); |
| props.put("bootstrap.servers", getPlainTextServiceUrl()); |
| props.put("group.id", "my-subscription-name"); |
| props.put("enable.auto.commit", "true"); |
| props.put("key.deserializer", StringDeserializer.class.getName()); |
| props.put("value.deserializer", StringDeserializer.class.getName()); |
| |
| // Create Kakfa consumer and verify all messages came from intended partition |
| @Cleanup |
| Consumer<String, String> consumer = new KafkaConsumer<>(props); |
| consumer.subscribe(Arrays.asList(topic)); |
| |
| int N = 8 * 3; |
| |
| final int choosenPartition = 5; |
| |
| for (int i = 0; i < N; i++) { |
| producer.send(new ProducerRecord<>(topic, choosenPartition, i, "hello-" + i)); |
| } |
| |
| producer.flush(); |
| |
| for (int i = 0; i < N;) { |
| ConsumerRecords<String, String> records = consumer.poll(100); |
| i += records.count(); |
| |
| records.forEach(record -> { |
| assertEquals(record.partition(), choosenPartition); |
| }); |
| } |
| |
| // No more messages for this consumer |
| ConsumerRecords<String, String> records = consumer.poll(100); |
| assertEquals(records.count(), 0); |
| } |
| |
| public static class MyCustomPartitioner implements Partitioner { |
| |
| static int USED_PARTITION = 3; |
| |
| @Override |
| public void configure(Map<String, ?> conf) { |
| // Do nothing |
| } |
| |
| @Override |
| public void close() { |
| // Do nothing |
| } |
| |
| @Override |
| public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { |
| // Dummy implementation that always return same partition |
| return USED_PARTITION; |
| } |
| } |
| |
| @Test |
| public void testCustomRouter() throws Exception { |
| String topic = "testCustomRouter"; |
| |
| // Create 8 partitions in topic |
| @Cleanup |
| PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(getHttpServiceUrl()).build(); |
| admin.topics().createPartitionedTopic(topic, 8); |
| |
| Properties producerProperties = new Properties(); |
| producerProperties.put("bootstrap.servers", getPlainTextServiceUrl()); |
| producerProperties.put("key.serializer", IntegerSerializer.class.getName()); |
| producerProperties.put("value.serializer", StringSerializer.class.getName()); |
| producerProperties.put("partitioner.class", MyCustomPartitioner.class.getName()); |
| |
| @Cleanup |
| Producer<Integer, String> producer = new KafkaProducer<>(producerProperties); |
| |
| Properties props = new Properties(); |
| props.put("bootstrap.servers", getPlainTextServiceUrl()); |
| props.put("group.id", "my-subscription-name"); |
| props.put("enable.auto.commit", "true"); |
| props.put("key.deserializer", IntegerDeserializer.class.getName()); |
| props.put("value.deserializer", StringDeserializer.class.getName()); |
| |
| // Create Kakfa consumer and verify all messages came from intended partition |
| @Cleanup |
| Consumer<Integer, String> consumer = new KafkaConsumer<>(props); |
| consumer.subscribe(Arrays.asList(topic)); |
| |
| int N = 8 * 3; |
| |
| for (int i = 0; i < N; i++) { |
| producer.send(new ProducerRecord<>(topic, i, "hello-" + i)); |
| } |
| |
| producer.flush(); |
| |
| for (int i = 0; i < N;) { |
| ConsumerRecords<Integer, String> records = consumer.poll(100); |
| i += records.count(); |
| |
| records.forEach(record -> { |
| assertEquals(record.partition(), MyCustomPartitioner.USED_PARTITION); |
| }); |
| } |
| |
| // No more messages for this consumer |
| ConsumerRecords<Integer, String> records = consumer.poll(100); |
| assertEquals(records.count(), 0); |
| } |
| |
| @Test |
| public void testConsumerSeek() throws Exception { |
| String topic = "testConsumerSeek"; |
| |
| Properties props = new Properties(); |
| props.put("bootstrap.servers", getPlainTextServiceUrl()); |
| props.put("group.id", "my-subscription-name"); |
| props.put("enable.auto.commit", "false"); |
| props.put("key.deserializer", StringDeserializer.class.getName()); |
| props.put("value.deserializer", StringDeserializer.class.getName()); |
| props.put("pulsar.consumer.acknowledgments.group.time.millis", "0"); |
| |
| Consumer<String, String> consumer = new KafkaConsumer<>(props); |
| consumer.subscribe(Arrays.asList(topic)); |
| |
| @Cleanup |
| PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPlainTextServiceUrl()).build(); |
| org.apache.pulsar.client.api.Producer<byte[]> pulsarProducer = pulsarClient.newProducer().topic(topic).create(); |
| |
| for (int i = 0; i < 10; i++) { |
| pulsarProducer.newMessage().key(Integer.toString(i)).value(("hello-" + i).getBytes()).send(); |
| } |
| |
| AtomicInteger received = new AtomicInteger(); |
| while (received.get() < 10) { |
| ConsumerRecords<String, String> records = consumer.poll(100); |
| records.forEach(record -> { |
| assertEquals(record.key(), Integer.toString(received.get())); |
| assertEquals(record.value(), "hello-" + received.get()); |
| |
| received.incrementAndGet(); |
| }); |
| |
| consumer.commitSync(); |
| } |
| |
| consumer.seekToBeginning(Collections.emptyList()); |
| |
| Thread.sleep(500); |
| |
| // Messages should be available again |
| received.set(0); |
| while (received.get() < 10) { |
| ConsumerRecords<String, String> records = consumer.poll(100); |
| records.forEach(record -> { |
| assertEquals(record.key(), Integer.toString(received.get())); |
| assertEquals(record.value(), "hello-" + received.get()); |
| |
| received.incrementAndGet(); |
| }); |
| |
| consumer.commitSync(); |
| } |
| |
| consumer.close(); |
| } |
| |
| @Test |
| public void testConsumerSeekToEnd() throws Exception { |
| String topic = "testConsumerSeekToEnd"; |
| |
| Properties props = new Properties(); |
| props.put("bootstrap.servers", getPlainTextServiceUrl()); |
| props.put("group.id", "my-subscription-name"); |
| props.put("enable.auto.commit", "false"); |
| props.put("key.deserializer", StringDeserializer.class.getName()); |
| props.put("value.deserializer", StringDeserializer.class.getName()); |
| props.put("pulsar.consumer.acknowledgments.group.time.millis", "0"); |
| |
| Consumer<String, String> consumer = new KafkaConsumer<>(props); |
| consumer.subscribe(Arrays.asList(topic)); |
| |
| @Cleanup |
| PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPlainTextServiceUrl()).build(); |
| org.apache.pulsar.client.api.Producer<byte[]> pulsarProducer = pulsarClient.newProducer().topic(topic).create(); |
| |
| for (int i = 0; i < 10; i++) { |
| pulsarProducer.newMessage().key(Integer.toString(i)).value(("hello-" + i).getBytes()).send(); |
| } |
| |
| AtomicInteger received = new AtomicInteger(); |
| while (received.get() < 10) { |
| ConsumerRecords<String, String> records = consumer.poll(100); |
| records.forEach(record -> { |
| assertEquals(record.key(), Integer.toString(received.get())); |
| assertEquals(record.value(), "hello-" + received.get()); |
| |
| received.incrementAndGet(); |
| }); |
| |
| consumer.commitSync(); |
| } |
| |
| consumer.seekToEnd(Collections.emptyList()); |
| Thread.sleep(500); |
| |
| consumer.close(); |
| |
| // Recreate the consumer |
| consumer = new KafkaConsumer<>(props); |
| consumer.subscribe(Arrays.asList(topic)); |
| |
| ConsumerRecords<String, String> records = consumer.poll(100); |
| // Since we are at the end of the topic, there should be no messages |
| assertEquals(records.count(), 0); |
| |
| consumer.close(); |
| } |
| |
| @Test |
| public void testSimpleProducer() throws Exception { |
| String topic = "testSimpleProducer"; |
| |
| @Cleanup |
| PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPlainTextServiceUrl()).build(); |
| org.apache.pulsar.client.api.Consumer<byte[]> pulsarConsumer = pulsarClient.newConsumer().topic(topic) |
| .subscriptionName("my-subscription") |
| .subscribe(); |
| |
| Properties props = new Properties(); |
| props.put("bootstrap.servers", getPlainTextServiceUrl()); |
| |
| props.put("key.serializer", IntegerSerializer.class.getName()); |
| props.put("value.serializer", StringSerializer.class.getName()); |
| |
| Producer<Integer, String> producer = new KafkaProducer<>(props); |
| |
| for (int i = 0; i < 10; i++) { |
| producer.send(new ProducerRecord<Integer, String>(topic, i, "hello-" + i)); |
| } |
| |
| producer.flush(); |
| producer.close(); |
| |
| for (int i = 0; i < 10; i++) { |
| Message<byte[]> msg = pulsarConsumer.receive(1, TimeUnit.SECONDS); |
| assertEquals(new String(msg.getData()), "hello-" + i); |
| pulsarConsumer.acknowledge(msg); |
| } |
| } |
| |
| @Test(timeOut = 10000) |
| public void testProducerCallback() throws Exception { |
| String topic = "testProducerCallback"; |
| |
| @Cleanup |
| PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPlainTextServiceUrl()).build(); |
| org.apache.pulsar.client.api.Consumer<byte[]> pulsarConsumer = pulsarClient.newConsumer() |
| .topic(topic) |
| .subscriptionName("my-subscription") |
| .subscribe(); |
| |
| Properties props = new Properties(); |
| props.put("bootstrap.servers", getPlainTextServiceUrl()); |
| |
| props.put("key.serializer", IntegerSerializer.class.getName()); |
| props.put("value.serializer", StringSerializer.class.getName()); |
| |
| Producer<Integer, String> producer = new KafkaProducer<>(props); |
| |
| CountDownLatch counter = new CountDownLatch(10); |
| |
| for (int i = 0; i < 10; i++) { |
| producer.send(new ProducerRecord<Integer, String>(topic, i, "hello-" + i), (metadata, exception) -> { |
| assertEquals(metadata.topic(), topic); |
| assertNull(exception); |
| |
| counter.countDown(); |
| }); |
| } |
| |
| counter.await(); |
| |
| for (int i = 0; i < 10; i++) { |
| Message<byte[]> msg = pulsarConsumer.receive(1, TimeUnit.SECONDS); |
| assertEquals(new String(msg.getData()), "hello-" + i); |
| pulsarConsumer.acknowledge(msg); |
| } |
| |
| producer.close(); |
| } |
| |
| @Test |
| public void testProducerAvroSchemaWithPulsarKafkaClient() throws Exception { |
| String topic = "testProducerAvroSchemaWithPulsarKafkaClient"; |
| AvroSchema<Bar> barSchema = AvroSchema.of(SchemaDefinition.<Bar>builder().withPojo(Bar.class).build()); |
| AvroSchema<Foo> fooSchema = AvroSchema.of(SchemaDefinition.<Foo>builder().withPojo(Foo.class).build()); |
| @Cleanup |
| PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPlainTextServiceUrl()).build(); |
| org.apache.pulsar.client.api.Consumer<byte[]> pulsarConsumer = |
| pulsarClient.newConsumer() |
| .topic(topic) |
| .subscriptionName("my-subscription") |
| .subscribe(); |
| Properties props = new Properties(); |
| props.put("bootstrap.servers", getPlainTextServiceUrl()); |
| props.put("key.serializer", IntegerSerializer.class.getName()); |
| props.put("value.serializer", StringSerializer.class.getName()); |
| |
| Producer<Bar, Foo> producer = new KafkaProducer<>(props, barSchema, fooSchema); |
| for (int i = 0; i < 10; i++) { |
| Bar bar = new Bar(); |
| bar.setField1(true); |
| |
| Foo foo = new Foo(); |
| foo.setField1("field1"); |
| foo.setField2("field2"); |
| foo.setField3(i); |
| producer.send(new ProducerRecord<Bar, Foo>(topic, bar, foo)); |
| } |
| producer.flush(); |
| producer.close(); |
| |
| for (int i = 0; i < 10; i++) { |
| Message<byte[]> msg = pulsarConsumer.receive(1, TimeUnit.SECONDS); |
| Foo value = fooSchema.decode(msg.getValue()); |
| Assert.assertEquals(value.getField1(), "field1"); |
| Assert.assertEquals(value.getField2(), "field2"); |
| Assert.assertEquals(value.getField3(), i); |
| pulsarConsumer.acknowledge(msg); |
| } |
| } |
| |
| @Test |
| public void testConsumerAvroSchemaWithPulsarKafkaClient() throws Exception { |
| String topic = "testConsumerAvroSchemaWithPulsarKafkaClient"; |
| |
| StringSchema stringSchema = new StringSchema(); |
| AvroSchema<Foo> fooSchema = AvroSchema.of(SchemaDefinition.<Foo>builder().withPojo(Foo.class).build()); |
| |
| Properties props = new Properties(); |
| props.put("bootstrap.servers", getPlainTextServiceUrl()); |
| props.put("group.id", "my-subscription-name"); |
| props.put("enable.auto.commit", "false"); |
| props.put("key.deserializer", StringDeserializer.class.getName()); |
| props.put("value.deserializer", StringDeserializer.class.getName()); |
| |
| @Cleanup |
| Consumer<String, Foo> consumer = new KafkaConsumer<String, Foo>(props, new StringSchema(), fooSchema); |
| consumer.subscribe(Arrays.asList(topic)); |
| |
| @Cleanup |
| PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPlainTextServiceUrl()).build(); |
| org.apache.pulsar.client.api.Producer<Foo> pulsarProducer = pulsarClient.newProducer(fooSchema).topic(topic).create(); |
| |
| for (int i = 0; i < 10; i++) { |
| Foo foo = new Foo(); |
| foo.setField1("field1"); |
| foo.setField2("field2"); |
| foo.setField3(i); |
| pulsarProducer.newMessage().keyBytes(stringSchema.encode(Integer.toString(i))).value(foo).send(); |
| } |
| |
| AtomicInteger received = new AtomicInteger(); |
| while (received.get() < 10) { |
| ConsumerRecords<String, Foo> records = consumer.poll(100); |
| if (!records.isEmpty()) { |
| records.forEach(record -> { |
| Assert.assertEquals(record.key(), Integer.toString(received.get())); |
| Foo value = record.value(); |
| Assert.assertEquals(value.getField1(), "field1"); |
| Assert.assertEquals(value.getField2(), "field2"); |
| Assert.assertEquals(value.getField3(), received.get()); |
| received.incrementAndGet(); |
| }); |
| |
| consumer.commitSync(); |
| } |
| } |
| } |
| |
| @Test |
| public void testProducerConsumerAvroSchemaWithPulsarKafkaClient() throws Exception { |
| String topic = "testProducerConsumerAvroSchemaWithPulsarKafkaClient"; |
| |
| AvroSchema<Bar> barSchema = AvroSchema.of(SchemaDefinition.<Bar>builder().withPojo(Bar.class).build()); |
| AvroSchema<Foo> fooSchema = AvroSchema.of(SchemaDefinition.<Foo>builder().withPojo(Foo.class).build()); |
| |
| Properties props = new Properties(); |
| props.put("bootstrap.servers", getPlainTextServiceUrl()); |
| props.put("group.id", "my-subscription-name"); |
| props.put("enable.auto.commit", "false"); |
| props.put("key.serializer", IntegerSerializer.class.getName()); |
| props.put("value.serializer", StringSerializer.class.getName()); |
| props.put("key.deserializer", StringDeserializer.class.getName()); |
| props.put("value.deserializer", StringDeserializer.class.getName()); |
| |
| @Cleanup |
| Consumer<Bar, Foo> consumer = new KafkaConsumer<>(props, barSchema, fooSchema); |
| consumer.subscribe(Arrays.asList(topic)); |
| |
| Producer<Bar, Foo> producer = new KafkaProducer<>(props, barSchema, fooSchema); |
| |
| for (int i = 0; i < 10; i++) { |
| Bar bar = new Bar(); |
| bar.setField1(true); |
| |
| Foo foo = new Foo(); |
| foo.setField1("field1"); |
| foo.setField2("field2"); |
| foo.setField3(i); |
| producer.send(new ProducerRecord<>(topic, bar, foo)); |
| } |
| producer.flush(); |
| producer.close(); |
| |
| AtomicInteger received = new AtomicInteger(); |
| while (received.get() < 10) { |
| ConsumerRecords<Bar, Foo> records = consumer.poll(100); |
| if (!records.isEmpty()) { |
| records.forEach(record -> { |
| Bar key = record.key(); |
| Assert.assertTrue(key.isField1()); |
| Foo value = record.value(); |
| Assert.assertEquals(value.getField1(), "field1"); |
| Assert.assertEquals(value.getField2(), "field2"); |
| Assert.assertEquals(value.getField3(), received.get()); |
| received.incrementAndGet(); |
| }); |
| |
| consumer.commitSync(); |
| } |
| } |
| } |
| |
| @Test |
| public void testProducerConsumerJsonSchemaWithPulsarKafkaClient() throws Exception { |
| String topic = "testProducerConsumerJsonSchemaWithPulsarKafkaClient"; |
| |
| JSONSchema<Bar> barSchema = JSONSchema.of(SchemaDefinition.<Bar>builder().withPojo(Bar.class).build()); |
| JSONSchema<Foo> fooSchema = JSONSchema.of(SchemaDefinition.<Foo>builder().withPojo(Foo.class).build()); |
| |
| Properties props = new Properties(); |
| props.put("bootstrap.servers", getPlainTextServiceUrl()); |
| props.put("group.id", "my-subscription-name"); |
| props.put("enable.auto.commit", "false"); |
| props.put("key.serializer", IntegerSerializer.class.getName()); |
| props.put("value.serializer", StringSerializer.class.getName()); |
| props.put("key.deserializer", StringDeserializer.class.getName()); |
| props.put("value.deserializer", StringDeserializer.class.getName()); |
| |
| @Cleanup |
| Consumer<Bar, Foo> consumer = new KafkaConsumer<>(props, barSchema, fooSchema); |
| consumer.subscribe(Arrays.asList(topic)); |
| |
| Producer<Bar, Foo> producer = new KafkaProducer<>(props, barSchema, fooSchema); |
| |
| for (int i = 0; i < 10; i++) { |
| Bar bar = new Bar(); |
| bar.setField1(true); |
| |
| Foo foo = new Foo(); |
| foo.setField1("field1"); |
| foo.setField2("field2"); |
| foo.setField3(i); |
| producer.send(new ProducerRecord<>(topic, bar, foo)); |
| } |
| producer.flush(); |
| producer.close(); |
| |
| AtomicInteger received = new AtomicInteger(); |
| while (received.get() < 10) { |
| ConsumerRecords<Bar, Foo> records = consumer.poll(100); |
| if (!records.isEmpty()) { |
| records.forEach(record -> { |
| Bar key = record.key(); |
| Assert.assertTrue(key.isField1()); |
| Foo value = record.value(); |
| Assert.assertEquals(value.getField1(), "field1"); |
| Assert.assertEquals(value.getField2(), "field2"); |
| Assert.assertEquals(value.getField3(), received.get()); |
| received.incrementAndGet(); |
| }); |
| |
| consumer.commitSync(); |
| } |
| } |
| } |
| |
| @Test |
| public void testProducerConsumerMixedSchemaWithPulsarKafkaClient() throws Exception { |
| String topic = "testProducerConsumerMixedSchemaWithPulsarKafkaClient"; |
| |
| Schema<String> keySchema = new PulsarKafkaSchema<>(new StringSerializer(), new StringDeserializer()); |
| JSONSchema<Foo> valueSchema = JSONSchema.of(SchemaDefinition.<Foo>builder().withPojo(Foo.class).build()); |
| |
| Properties props = new Properties(); |
| props.put("bootstrap.servers", getPlainTextServiceUrl()); |
| props.put("group.id", "my-subscription-name"); |
| props.put("enable.auto.commit", "false"); |
| props.put("key.serializer", IntegerSerializer.class.getName()); |
| props.put("value.serializer", StringSerializer.class.getName()); |
| props.put("key.deserializer", StringDeserializer.class.getName()); |
| props.put("value.deserializer", StringDeserializer.class.getName()); |
| |
| @Cleanup |
| Consumer<String, Foo> consumer = new KafkaConsumer<>(props, keySchema, valueSchema); |
| consumer.subscribe(Arrays.asList(topic)); |
| |
| Producer<String, Foo> producer = new KafkaProducer<>(props, keySchema, valueSchema); |
| |
| for (int i = 0; i < 10; i++) { |
| Foo foo = new Foo(); |
| foo.setField1("field1"); |
| foo.setField2("field2"); |
| foo.setField3(i); |
| producer.send(new ProducerRecord<>(topic, "hello" + i, foo)); |
| } |
| producer.flush(); |
| producer.close(); |
| |
| AtomicInteger received = new AtomicInteger(); |
| while (received.get() < 10) { |
| ConsumerRecords<String, Foo> records = consumer.poll(100); |
| if (!records.isEmpty()) { |
| records.forEach(record -> { |
| String key = record.key(); |
| Assert.assertEquals(key, "hello" + received.get()); |
| Foo value = record.value(); |
| Assert.assertEquals(value.getField1(), "field1"); |
| Assert.assertEquals(value.getField2(), "field2"); |
| Assert.assertEquals(value.getField3(), received.get()); |
| received.incrementAndGet(); |
| }); |
| |
| consumer.commitSync(); |
| } |
| } |
| } |
| } |
