From f6d98025ee5eb6bb3181e83193e33bfbdf0cb963 Mon Sep 17 00:00:00 2001 From: Kevin Wilson Date: Thu, 15 Apr 2021 15:17:42 -0600 Subject: [PATCH 1/2] Add support for deduplication testing --- pulsar-testclient/pom.xml | 7 ++ .../testclient/DefaultMessageFormatter.java | 115 ++++++++++++++++++ .../pulsar/testclient/IMessageFormatter.java | 23 ++++ .../testclient/PerformanceProducer.java | 53 +++++++- .../testclient/PerformanceProducerTest.java | 13 ++ .../TestDefaultMessageFormatter.java | 63 ++++++++++ 6 files changed, 270 insertions(+), 4 deletions(-) create mode 100644 pulsar-testclient/src/main/java/org/apache/pulsar/testclient/DefaultMessageFormatter.java create mode 100644 pulsar-testclient/src/main/java/org/apache/pulsar/testclient/IMessageFormatter.java create mode 100644 pulsar-testclient/src/test/java/org/apache/pulsar/testclient/TestDefaultMessageFormatter.java diff --git a/pulsar-testclient/pom.xml b/pulsar-testclient/pom.xml index a244e15f94237..e572abfc0b66a 100644 --- a/pulsar-testclient/pom.xml +++ b/pulsar-testclient/pom.xml @@ -94,6 +94,13 @@ org.hdrhistogram HdrHistogram + + org.json + json + 20201115 + test + + diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/DefaultMessageFormatter.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/DefaultMessageFormatter.java new file mode 100644 index 0000000000000..a84fb2d7e6f4d --- /dev/null +++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/DefaultMessageFormatter.java @@ -0,0 +1,115 @@ +/** + * 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.testclient; + +import java.nio.charset.StandardCharsets; +import java.util.Random; + +public class DefaultMessageFormatter implements IMessageFormatter { + Random r = new Random(); + + + @Override + public byte[] formatMessage(String producerName, long msgId, byte[] message) { + String sMessage = new String(message, StandardCharsets.UTF_8); + if (producerName != null && !producerName.isEmpty()) { + sMessage = sMessage.replaceAll("%p", producerName); + } + sMessage = sMessage.replaceAll("%i", String.valueOf(msgId)); + sMessage = sMessage.replaceAll("%t", String.valueOf(System.nanoTime())); + + int idx = sMessage.indexOf("%"); + while (idx > 0) { + + float size = 0; + int i=1; + for (; idx+i < sMessage.length(); i++) { + char c = sMessage.charAt(idx + i); + if (Character.isDigit(c) && c != '.') { + continue; + } + if (c == '.' || c == '-') { + continue; + } + break; + } + if (i != 1) { + size = Float.valueOf(new String(sMessage.substring(idx+1,idx+i))); + } + + String sub = sMessage.substring(idx, idx+i+1); + + if (sMessage.charAt(idx+i) == 'f') { + sMessage=sMessage.replaceFirst(sub, getFloatValue(size)); + } else if (sMessage.charAt(idx+i) == 'l') { + sMessage=sMessage.replaceFirst(sub, getLongValue(size)); + } else if (sMessage.charAt(idx+i) == 'd') { + sMessage = sMessage.replaceFirst(sub, getIntValue(size)); + } else if (sMessage.charAt(idx+i) == 's') { + sMessage = sMessage.replaceFirst(sub, getStringValue(size)); + } + idx = sMessage.indexOf("%", idx); + } + return sMessage.getBytes(StandardCharsets.UTF_8); + } + + private float _getFloatValue(float size) { + float f = r.nextFloat(); + int mag = (int) Math.abs(size); + f = f * (float) Math.pow(10, mag); + if (size < 0 && ((int) f) % 2 == 1) { + return f * -1; + } + return f; + } + + private String getStringValue(float size) { + int s = (int) size; + if (size == 0) { + size = 20; + }; + String result = ""; + for(int i = 0; i < s; i++) { + result = result + (char) ((int) 'a' + (int) (r.nextFloat() * 26)); + } + return result; + } + + private String getFloatValue(float size) { + if (size == 0) { + return String.valueOf(r.nextFloat()); + } + String format = "%" + String.valueOf(size) + "f"; + + return String.format(format, _getFloatValue(size)); + } + + private String getIntValue(float size) { + if (size == 0) { + return String.valueOf(r.nextInt()); + } + return String.valueOf((int) _getFloatValue(size)); + } + private String getLongValue(float size) { + if (size == 0) { + return String.valueOf(r.nextLong()); + } + return String.valueOf((long) _getFloatValue(size)); + } +} diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/IMessageFormatter.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/IMessageFormatter.java new file mode 100644 index 0000000000000..e80ae50484c27 --- /dev/null +++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/IMessageFormatter.java @@ -0,0 +1,23 @@ +/** + * 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.testclient; + +public interface IMessageFormatter { + byte[] formatMessage(String producerName, long msg, byte[] message); +} \ No newline at end of file diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java index 5ab4bb058fbd0..a75910e40cc67 100644 --- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java +++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java @@ -93,6 +93,8 @@ public class PerformanceProducer { private static Recorder recorder = new Recorder(TimeUnit.SECONDS.toMicros(120000), 5); private static Recorder cumulativeRecorder = new Recorder(TimeUnit.SECONDS.toMicros(120000), 5); + private static IMessageFormatter messageFormatter = null; + static class Arguments { @Parameter(names = { "-h", "--help" }, description = "Help message", help = true) @@ -119,6 +121,15 @@ static class Arguments { @Parameter(names = { "-n", "--num-producers" }, description = "Number of producers (per topic)") public int numProducers = 1; + @Parameter(names = {"--separator"}, description = "Separator between the topic and topic number") + public String separator = "-"; + + @Parameter(names = {"--send-timeout"}, description = "Set the sendTimeout value default 0 to keep compatibility with previous version of pulsar-perf") + public int sendTimeout = 0; + + @Parameter(names = { "-pn", "--producer-name" }, description = "Producer Name") + public String producerName = null; + @Parameter(names = { "-u", "--service-url" }, description = "Pulsar Service URL") public String serviceURL; @@ -227,6 +238,13 @@ static class Arguments { @Parameter(names = { "-am", "--access-mode" }, description = "Producer access mode") public ProducerAccessMode producerAccessMode = ProducerAccessMode.Shared; + + @Parameter(names = { "-fp", "--format-payload" }, + description = "Format %i as a message index in the stream from producer and/or %t as the timestamp nanoseconds.") + public boolean formatPayload = false; + + @Parameter(names = {"-fc", "--format-class"}, description="Custom Formatter class name") + public String formatterClass = "org.apache.pulsar.testclient.DefaultMessageFormatter"; } public static void main(String[] args) throws Exception { @@ -254,7 +272,7 @@ public static void main(String[] args) throws Exception { String prefixTopicName = arguments.topics.get(0); List defaultTopics = Lists.newArrayList(); for (int i = 0; i < arguments.numTopics; i++) { - defaultTopics.add(String.format("%s-%d", prefixTopicName, i)); + defaultTopics.add(String.format("%s%s%d", prefixTopicName, arguments.separator, i)); } arguments.topics = defaultTopics; } else { @@ -330,6 +348,10 @@ public static void main(String[] args) throws Exception { for (String payload : payloadList) { payloadByteList.add(payload.getBytes(StandardCharsets.UTF_8)); } + + if (arguments.formatPayload) { + messageFormatter = getMessageFormatter(arguments.formatterClass); + } } else { for (int i = 0; i < payloadBytes.length; ++i) { payloadBytes[i] = (byte) (random.nextInt(26) + 65); @@ -386,6 +408,7 @@ public static void main(String[] args) throws Exception { executor.submit(() -> { log.info("Started performance test thread {}", threadIdx); runProducer( + threadIdx, arguments, numMessagesPerThread, msgRatePerThread, @@ -451,7 +474,18 @@ public static void main(String[] args) throws Exception { } } - private static void runProducer(Arguments arguments, + static IMessageFormatter getMessageFormatter(String formatterClass) { + try { + ClassLoader classLoader = PerformanceProducer.class.getClassLoader(); + Class clz = classLoader.loadClass(formatterClass); + return (IMessageFormatter) clz.newInstance(); + } catch (Exception e) { + return null; + } + } + + private static void runProducer(int producerId, + Arguments arguments, long numMessages, int msgRate, List payloadByteList, @@ -484,7 +518,7 @@ private static void runProducer(Arguments arguments, client = clientBuilder.build(); ProducerBuilder producerBuilder = client.newProducer() // - .sendTimeout(0, TimeUnit.SECONDS) // + .sendTimeout(arguments.sendTimeout, TimeUnit.SECONDS) // .compressionType(arguments.compression) // .maxPendingMessages(arguments.maxOutstanding) // .maxPendingMessagesAcrossPartitions(arguments.maxPendingMessagesAcrossPartitions) @@ -492,6 +526,11 @@ private static void runProducer(Arguments arguments, // enable round robin message routing if it is a partitioned topic .messageRoutingMode(MessageRoutingMode.RoundRobinPartition); + if (arguments.producerName != null) { + String producerName = String.format("%s%s%d", arguments.producerName, arguments.separator, producerId); + producerBuilder.producerName(producerName); + } + if (arguments.batchTimeMillis == 0.0 && arguments.batchMaxMessages == 0) { producerBuilder.enableBatching(false); } else { @@ -514,6 +553,7 @@ private static void runProducer(Arguments arguments, } for (int i = 0; i < arguments.numTopics; i++) { + String topic = arguments.topics.get(i); log.info("Adding {} publishers on topic {}", arguments.numProducers, topic); @@ -578,7 +618,12 @@ private static void runProducer(Arguments arguments, byte[] payloadData; if (arguments.payloadFilename != null) { - payloadData = payloadByteList.get(random.nextInt(payloadByteList.size())); + if (messageFormatter != null) { + payloadData = messageFormatter.formatMessage(arguments.producerName, totalSent, + payloadByteList.get(random.nextInt(payloadByteList.size()))); + } else { + payloadData = payloadByteList.get(random.nextInt(payloadByteList.size())); + } } else { payloadData = payloadBytes; } diff --git a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceProducerTest.java b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceProducerTest.java index 716a05a258e80..01588c62743c1 100644 --- a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceProducerTest.java +++ b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceProducerTest.java @@ -176,4 +176,17 @@ public void testCreatePartitions() throws Exception { thread.join(); Assert.assertEquals(10, pulsar.getAdminClient().topics().getPartitionedTopicMetadata(topic).partitions); } + + @Test + public void testNotExistIMessageFormatter() { + IMessageFormatter msgFormatter = PerformanceProducer.getMessageFormatter("org.apache.pulsar.testclient.NonExistentFormatter"); + Assert.assertNull(msgFormatter); + } + + @Test + public void testDefaultIMessageFormatter() { + IMessageFormatter msgFormatter = PerformanceProducer.getMessageFormatter("org.apache.pulsar.testclient.DefaultMessageFormatter"); + Assert.assertTrue(msgFormatter instanceof DefaultMessageFormatter); + } + } diff --git a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/TestDefaultMessageFormatter.java b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/TestDefaultMessageFormatter.java new file mode 100644 index 0000000000000..32cb08f449416 --- /dev/null +++ b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/TestDefaultMessageFormatter.java @@ -0,0 +1,63 @@ +/** + * 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.testclient; + +import org.testng.Assert; +import org.testng.annotations.Test; +import org.json.JSONObject; + +import java.nio.charset.StandardCharsets; + +public class TestDefaultMessageFormatter { + + @Test + public void testFormatMessage() { + String producerName = "producer-1"; + long msgId = 3; + byte[] message = "{ \"producer\": \"%p\", \"msgId\": %i, \"nanoTime\": %t, \"float1\": %5.2f, \"float2\": %-5.2f, \"long1\": %12l, \"long2\": %l, \"int1\": %d, \"int2\": %1d , \"long3\": %5l, \"str\": \"%5s\" }".getBytes(); + byte[] formatted = new DefaultMessageFormatter().formatMessage(producerName, msgId, message); + String jsonString = new String(formatted, StandardCharsets.UTF_8); + + JSONObject obj = new JSONObject(jsonString); + String prod = obj.getString("producer"); + int mid = obj.getInt("msgId"); + long nt = obj.getLong("nanoTime"); + float f1 = obj.getFloat("float1"); + float f2 = obj.getFloat("float2"); + long l1 = obj.getLong("long1"); + long l2 = obj.getLong("long2"); + long i1 = obj.getLong("int1"); + long i2 = obj.getLong("int2"); + String str = obj.getString("str"); + long l3 = obj.getLong("long3"); + Assert.assertEquals(producerName, prod); + Assert.assertEquals(msgId, mid); + Assert.assertTrue( nt > 0); + Assert.assertNotEquals(f1, f2); + Assert.assertNotEquals(l1, l2); + Assert.assertNotEquals(i1, i2); + Assert.assertTrue(l3 > 0); + Assert.assertTrue(l3 <= 99999); + Assert.assertTrue(i2 < 10); + Assert.assertTrue(0 < i2); + Assert.assertTrue(f2 < 100000); + Assert.assertTrue( -100000 < f2); + + } +} From 636306a034ee97feb2a09b575f8bc05ce3b07fb6 Mon Sep 17 00:00:00 2001 From: Kevin Wilson Date: Fri, 16 Apr 2021 07:57:18 -0600 Subject: [PATCH 2/2] Move to fasterxml.jackson for json parsing in test case. --- pulsar-testclient/pom.xml | 7 ++-- .../TestDefaultMessageFormatter.java | 37 ++++++++++++------- 2 files changed, 27 insertions(+), 17 deletions(-) diff --git a/pulsar-testclient/pom.xml b/pulsar-testclient/pom.xml index e572abfc0b66a..026c1cca7aae8 100644 --- a/pulsar-testclient/pom.xml +++ b/pulsar-testclient/pom.xml @@ -94,11 +94,10 @@ org.hdrhistogram HdrHistogram + - org.json - json - 20201115 - test + com.fasterxml.jackson.core + jackson-databind diff --git a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/TestDefaultMessageFormatter.java b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/TestDefaultMessageFormatter.java index 32cb08f449416..2d5c4b9a3bfbc 100644 --- a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/TestDefaultMessageFormatter.java +++ b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/TestDefaultMessageFormatter.java @@ -18,11 +18,13 @@ */ package org.apache.pulsar.testclient; +import com.fasterxml.jackson.core.JsonParseException; +import com.fasterxml.jackson.databind.JsonNode; import org.testng.Assert; import org.testng.annotations.Test; -import org.json.JSONObject; import java.nio.charset.StandardCharsets; +import com.fasterxml.jackson.databind.ObjectMapper; public class TestDefaultMessageFormatter { @@ -34,18 +36,27 @@ public void testFormatMessage() { byte[] formatted = new DefaultMessageFormatter().formatMessage(producerName, msgId, message); String jsonString = new String(formatted, StandardCharsets.UTF_8); - JSONObject obj = new JSONObject(jsonString); - String prod = obj.getString("producer"); - int mid = obj.getInt("msgId"); - long nt = obj.getLong("nanoTime"); - float f1 = obj.getFloat("float1"); - float f2 = obj.getFloat("float2"); - long l1 = obj.getLong("long1"); - long l2 = obj.getLong("long2"); - long i1 = obj.getLong("int1"); - long i2 = obj.getLong("int2"); - String str = obj.getString("str"); - long l3 = obj.getLong("long3"); + ObjectMapper objectMapper = new ObjectMapper(); + + JsonNode obj = null; + try { + obj = objectMapper.readValue(jsonString, JsonNode.class); + + } catch(Exception jpe) { + Assert.fail("Exception parsing json"); + } + + String prod = obj.get("producer").asText(); + int mid = obj.get("msgId").asInt(); + long nt = obj.get("nanoTime").asLong(); + float f1 = obj.get("float1").floatValue(); + float f2 = obj.get("float2").floatValue(); + long l1 = obj.get("long1").asLong(); + long l2 = obj.get("long2").asLong(); + long i1 = obj.get("int1").asInt(); + long i2 = obj.get("int2").asInt(); + String str = obj.get("str").asText(); + long l3 = obj.get("long3").asLong(); Assert.assertEquals(producerName, prod); Assert.assertEquals(msgId, mid); Assert.assertTrue( nt > 0);