diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/IncrementedNumberMessageFormatter.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/IncrementedNumberMessageFormatter.java new file mode 100644 index 0000000000000..8d65a66ff157d --- /dev/null +++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/IncrementedNumberMessageFormatter.java @@ -0,0 +1,151 @@ +/* + * 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.io.File; +import java.io.FileInputStream; +import java.io.FileOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.LinkedList; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * message formatter used to generate long number from 0. + * For non-txn test, IncrementedNumberMessageFormatter support for resending messages + * when send failed. + * For txn tset, IncrementedNumberMessageFormatter support for taking snapshot of current + * Long value when the perf process is killed and restoring Long value from snapshot when + * the perf process is started; support for persisting messages in transaction to local files + * in case of txn abortion or process shutdown. + */ +public class IncrementedNumberMessageFormatter implements IMessageFormatter{ + private static final Logger log = LoggerFactory.getLogger(IncrementedNumberMessageFormatter.class); + + private AtomicLong atomicLong = new AtomicLong(); + + // tmp files corresponding to txn aborted. + public ConcurrentMap> tmpFilesWithAbortedTxn = new ConcurrentHashMap(); + + // blocking queue used for resending messages in non-txn produce. + public BlockingQueue msgToResend = new LinkedBlockingQueue<>(); + + private static ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES); + + private static final String snapshotFilePath = "snapshot.data"; + + public static synchronized byte[] longToBytes(long x) { + buffer.putLong(0, x); + return Arrays.copyOf(buffer.array(), Long.BYTES); + } + + public static synchronized long bytesToLong(byte[] bytes) { + buffer.clear(); // need clear to put + buffer.put(bytes, 0, bytes.length); + buffer.flip(); //need flip to get + return buffer.getLong(); + } + + @Override + public byte[] formatMessage(String producerName, long msg, byte[] message) { + return longToBytes(atomicLong.getAndAdd(1L)); + } + + public byte[] formatMessage() throws InterruptedException { + if (msgToResend.isEmpty()) { + return longToBytes(atomicLong.getAndAdd(1L)); + } else { + byte[] arr = msgToResend.poll(2, TimeUnit.SECONDS); + if (arr == null) { + return longToBytes(atomicLong.getAndAdd(1L)); + } else { + return arr; + } + } + } + + public byte[] formatMessage(File file) { + if (tmpFilesWithAbortedTxn.containsKey(file)) { + byte[] arr = tmpFilesWithAbortedTxn.get(file).removeFirst(); + if (tmpFilesWithAbortedTxn.get(file).size() == 0) { + tmpFilesWithAbortedTxn.remove(file); + } + return arr; + } + return formatMessage(null, 0L, null); + } + + public int registerTmpFileWithAbortedTxn(File file) throws IOException { + // read content persisted in tmp file. + FileInputStream fileInputStream = new FileInputStream(file); + byte[] arr = new byte[Long.BYTES]; + LinkedList tmpContent = new LinkedList<>(); + while (fileInputStream.read(arr) != -1) { + tmpContent.addLast(arr); + arr = new byte[Long.BYTES]; + } + fileInputStream.close(); + if (tmpContent.size() != 0) { + tmpFilesWithAbortedTxn.put(file, tmpContent); + } + return tmpContent.size(); + } + + public boolean takeSnapshot(){ + try { + FileOutputStream fileOutputStream = new FileOutputStream(snapshotFilePath); + fileOutputStream.write(longToBytes(atomicLong.get())); + fileOutputStream.close(); + log.info("Message Formatter take snapshot successfully. value:{}", atomicLong.get()); + return true; + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + public boolean recoverFromSnapshot() { + try { + File file = new File(snapshotFilePath); + if (file.exists()) { + FileInputStream fileInputStream = new FileInputStream(file); + byte[] arr = new byte[Long.BYTES]; + fileInputStream.read(arr); + atomicLong.compareAndSet(0, bytesToLong(arr)); + log.info("Message Formatter recover from snapshot file. value:{}", atomicLong.get()); + fileInputStream.close(); + return true; + } + } catch (IOException e) { + throw new RuntimeException(e); + } + return false; + } + + public long getMessageCount() { + return atomicLong.get(); + } +} 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 9aedc8c4bd327..52441cbc842c1 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 @@ -32,8 +32,12 @@ import com.fasterxml.jackson.databind.ObjectWriter; import com.google.common.util.concurrent.RateLimiter; import io.netty.util.concurrent.DefaultThreadFactory; +import java.io.File; +import java.io.FileNotFoundException; import java.io.FileOutputStream; +import java.io.IOException; import java.io.PrintStream; +import java.io.PrintWriter; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; @@ -44,6 +48,10 @@ import java.util.List; import java.util.Properties; import java.util.Random; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -57,6 +65,7 @@ import org.HdrHistogram.Histogram; import org.HdrHistogram.HistogramLogWriter; import org.HdrHistogram.Recorder; +import org.apache.commons.lang3.tuple.Pair; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.admin.PulsarAdminBuilder; import org.apache.pulsar.client.admin.PulsarAdminException; @@ -82,6 +91,19 @@ public class PerformanceProducer { private static final ExecutorService executor = Executors .newCachedThreadPool(new DefaultThreadFactory("pulsar-perf-producer-exec")); + /** + * used for concurrency control to tmp files. when a txn is created, a tmp file will + * be dequeued for messages persistence. when a txn is committed successfully, the + * corresponding file will be enqueued to tmpFiles. when a txn is aborted or timeout, + * the corresponding file not only will be put back into tmpFiles, but also put into + * MessageFormatter to record that this file have something need to be resent to broker + * in a new txn. + */ + private static BlockingQueue tmpFiles = null; + + // record which tmp file specific txn use. when specific txn is aborted, the corresponding file will be put into + private static ConcurrentMap> txnTmpFileMap = null; + private static final LongAdder messagesSent = new LongAdder(); private static final LongAdder messagesFailed = new LongAdder(); @@ -102,6 +124,9 @@ public class PerformanceProducer { private static IMessageFormatter messageFormatter = null; + // set true on exit, to avoid new transaction creation. + private static volatile boolean onExit = false; + @Parameters(commandDescription = "Test pulsar producer performance.") static class Arguments extends PerformanceBaseArguments { @@ -251,6 +276,25 @@ static class Arguments extends PerformanceBaseArguments { + "setting to true, -abort takes effect)") public boolean isAbortTransaction = false; + @Parameter(names = {"--txn-test-enabled"}, description = "Enable or disable the transaction consistency test") + public boolean isEnableTxnTest = false; + + @Parameter(names = {"--resend-on-failure"}, description = "resend message on failure in non-txn test") + public boolean resendOnFailure = false; + + @Parameter(names = "--retry-times", description = "the retry times when commit operation receive " + + "PulsarClientException$TimeoutException") + public int retryTimes = 3; + + @Parameter(names = "--base-dir-save-resend", description = "save the data resend in transaction, empty string" + + "to indicate that do not save resend data") + public String baseDirToSaveResendTxnData = ""; + + @Parameter(names = {"--max-txn"}, description = "max transactions per second, " + + "determining the number of tmp files") + public int maxTransactionsPerSecond = 20; + + @Parameter(names = { "--histogram-file" }, description = "HdrHistogram output file") public String histogramFile = null; @@ -353,11 +397,31 @@ public static void main(String[] args) throws Exception { } } + if (arguments.formatterClass.equals + ("org.apache.pulsar.testclient.IncrementedNumberMessageFormatter")) { + messageFormatter = getMessageFormatter(arguments.formatterClass); + if (arguments.isEnableTxnTest) { + IncrementedNumberMessageFormatter incrementedNumberMessageFormatter = + (IncrementedNumberMessageFormatter) messageFormatter; + incrementedNumberMessageFormatter.recoverFromSnapshot(); + if (arguments.numMessages > 0) { + arguments.numMessages -= incrementedNumberMessageFormatter.getMessageCount(); + } + } + } + + initializeTmpFiles(arguments); + long start = System.nanoTime(); Runtime.getRuntime().addShutdownHook(new Thread(() -> { printAggregatedThroughput(start, arguments); printAggregatedStats(); + handleTxnOnExit(arguments); + if (messageFormatter instanceof IncrementedNumberMessageFormatter + && arguments.isEnableTxnTest) { + ((IncrementedNumberMessageFormatter) messageFormatter).takeSnapshot(); + } })); if (arguments.partitions != null) { @@ -481,6 +545,195 @@ public static void main(String[] args) throws Exception { } } + static void initializeTmpFiles(Arguments arguments) throws InterruptedException, IOException { + if (arguments.isEnableTxnTest && arguments.isEnableTransaction + && arguments.maxTransactionsPerSecond > 0) { + // initialize tmpFiles + File tmpDataDir = new File("tmpData"); + if (!tmpDataDir.exists()) { + tmpDataDir.mkdir(); + } + tmpFiles = new ArrayBlockingQueue(arguments.maxTransactionsPerSecond); + for (int i = 0; i < arguments.maxTransactionsPerSecond; i++) { + File file = new File(tmpDataDir + "/tmp" + i + ".data"); + if (!file.exists()) { + try { + file.createNewFile(); + } catch (IOException e) { + System.out.println("can not create tmp files!"); + throw new RuntimeException(e); + } + } + tmpFiles.put(file); + // if there are any content in tmpFile, we need to register it into MessageFormatter + if (messageFormatter instanceof IncrementedNumberMessageFormatter) { + try { + int messagesCount = ((IncrementedNumberMessageFormatter) messageFormatter). + registerTmpFileWithAbortedTxn(file); + arguments.numMessages += messagesCount; + } catch (IOException e) { + throw new RuntimeException(e); + } + } + } + txnTmpFileMap = new ConcurrentHashMap<>(); + + // create dir for saving txn data, and clean up old data. + if (!arguments.baseDirToSaveResendTxnData.isEmpty()) { + File baseDir = new File(arguments.baseDirToSaveResendTxnData); + if (!baseDir.exists()) { + baseDir.mkdir(); + } + File[] files = baseDir.listFiles(); + if (files == null) { + throw new IOException("can not create baseDirToSaveResendTxnData"); + } else if (files.length != 0) { + for (File file : files) { + file.delete(); + } + } + } + } + } + + /** + * handle those transactions that are committed or aborted but corresponding tmp file + * do not clean up due to process shutdown. executed in hook thread. + * @param arguments + */ + static void handleTxnOnExit(Arguments arguments) { + if (arguments.isEnableTxnTest) { + // clean up tmpFiles to avoid new transaction be created. but ongoing + // txn may put tmp file back into tmpFiles. So we use onExit to determine + // whether put tmp file back into tmpFiles. + onExit = true; + tmpFiles.clear(); + + boolean needToWaitSomeTime = false; + long start = System.currentTimeMillis(); + for (Transaction txn : txnTmpFileMap.keySet()) { + Transaction.State state = txn.getState(); + if (state.equals(Transaction.State.COMMITTED) + || state.equals(Transaction.State.ABORTED) + || state.equals(Transaction.State.ERROR) + || state.equals(Transaction.State.TIME_OUT)) { + File tmpFile = txnTmpFileMap.get(txn).getLeft(); + if (tmpFile.exists()) { + // txn has been terminated, but the corresponding + // tmp file do not clean up. + tmpFile.delete(); + } + } else if (state.equals(Transaction.State.COMMITTING) + || state.equals(Transaction.State.ABORTING)) { + needToWaitSomeTime = true; + } + } + if (needToWaitSomeTime) { + while (!txnTmpFileMap.isEmpty()) { + // sleep for some time for executing asynchronous tasks completed. + try { + log.info("txnTmpFileMap size:{}", txnTmpFileMap.size()); + Thread.sleep(1000); + } catch (InterruptedException e) { + log.error("sleep failed! inconsistent situation occurs!"); + throw new RuntimeException(e); + } + } + } + } + } + + /** + * handle committed txn. close the corresponding tmp file, and clear + * out of the content of tmp file. + * @param txn + */ + static void handleTxnOnCommitted(Transaction txn) { + Pair pair = txnTmpFileMap.remove(txn); + try { + pair.getRight().close(); + clearContentsOfFile(pair.getLeft()); + if (!onExit) { + tmpFiles.put(pair.getLeft()); + } + } catch (IOException | InterruptedException e) { + throw new RuntimeException(e); + } + } + + /** + * handle aborted txn. close the corresponding tmp file, and register the content + * ot tmp file into IMessageFormatter + * + * @param txn + * @param messageFormatter + * @return the number of messages in aborted txn. + */ + static int handleTxnOnAborted(Transaction txn, IMessageFormatter messageFormatter, String baseDirToSaveResendData) { + Pair pair = txnTmpFileMap.remove(txn); + int messageCount = 0; + try { + pair.getRight().close(); + if (!baseDirToSaveResendData.isEmpty()) { + Files.copy(pair.getLeft().toPath(), new File(baseDirToSaveResendData + "/" + + txn.getTxnID().getMostSigBits() + ":" + txn.getTxnID().getLeastSigBits()).toPath()); + } + if (messageFormatter instanceof IncrementedNumberMessageFormatter) { + messageCount = ((IncrementedNumberMessageFormatter) messageFormatter). + registerTmpFileWithAbortedTxn(pair.getLeft()); + } + if (!onExit) { + tmpFiles.put(pair.getLeft()); + } + } catch (IOException | InterruptedException e) { + throw new RuntimeException(e); + } + return messageCount; + } + + /** + * clear out the content of file without deleting file. + * @param file + * @throws FileNotFoundException + */ + static void clearContentsOfFile(File file) throws FileNotFoundException { + PrintWriter writer = new PrintWriter(file); + writer.print(""); + writer.close(); + } + + /** + * check if file is empty, that is whether file.length()==0. + * @param file + * @return + */ + static boolean isEmpty(File file) { + if (file.exists()) { + return file.length() == 0; + } else { + try { + file.createNewFile(); + } catch (IOException e) { + throw new RuntimeException(e); + } + return true; + } + } + + static void openTmpFileStreamForNewTxn(File tmpFile, Transaction transaction) throws FileNotFoundException { + if (isEmpty(tmpFile)) { + txnTmpFileMap.put(transaction, Pair.of(tmpFile, new FileOutputStream(tmpFile, false))); + } else { + // there are content in tmpFile written by another txn, we can't overwrite it. + txnTmpFileMap.put(transaction, Pair.of(tmpFile, new FileOutputStream(tmpFile, true))); + } + } + + static boolean isTimeOutException(Throwable throwable) { + return throwable.getMessage().contains("Could not get" + + " response from transaction meta store within given timeout."); + } + static IMessageFormatter getMessageFormatter(String formatterClass) { try { ClassLoader classLoader = PerformanceProducer.class.getClassLoader(); @@ -559,6 +812,10 @@ private static void runProducer(int producerId, .withTransactionTimeout(arguments.transactionTimeout, TimeUnit.SECONDS) .build() .get()); + if (arguments.isEnableTxnTest) { + File tmpFile = tmpFiles.take(); + openTmpFileStreamForNewTxn(tmpFile, transactionAtomicReference.get()); + } } else { transactionAtomicReference = new AtomicReference<>(null); } @@ -600,7 +857,7 @@ private static void runProducer(int producerId, } } // Send messages on all topics/producers - long totalSent = 0; + final LongAdder totalSent = new LongAdder(); AtomicLong numMessageSend = new AtomicLong(0); Semaphore numMsgPerTxnLimit = new Semaphore(arguments.numMessagesPerTransaction); while (true) { @@ -616,13 +873,14 @@ private static void runProducer(int producerId, } if (numMessages > 0) { - if (totalSent++ >= numMessages) { + if (totalSent.longValue() >= numMessages) { log.info("------------- DONE (reached the maximum number: {} of production) --------------" , numMessages); doneLatch.countDown(); Thread.sleep(5000); PerfClientUtils.exit(0); } + totalSent.increment(); } rateLimiter.acquire(); //if transaction is disable, transaction will be null. @@ -633,15 +891,34 @@ private static void runProducer(int producerId, if (arguments.payloadFilename != null) { if (messageFormatter != null) { - payloadData = messageFormatter.formatMessage(arguments.producerName, totalSent, + payloadData = messageFormatter.formatMessage(arguments.producerName, totalSent.longValue(), payloadByteList.get(ThreadLocalRandom.current().nextInt(payloadByteList.size()))); } else { payloadData = payloadByteList.get( ThreadLocalRandom.current().nextInt(payloadByteList.size())); } + } else if (messageFormatter instanceof IncrementedNumberMessageFormatter) { + IncrementedNumberMessageFormatter incrementedNumberMessageFormatter = + (IncrementedNumberMessageFormatter) messageFormatter; + if (arguments.isEnableTxnTest) { + // txn test + Pair pair = txnTmpFileMap.get(transaction); + if (incrementedNumberMessageFormatter.tmpFilesWithAbortedTxn.containsKey(pair.getLeft())) { + // some messages in aborted txn are persisted in this file, we need to resend them. + payloadData = incrementedNumberMessageFormatter.formatMessage(pair.getLeft()); + } else { + payloadData = messageFormatter.formatMessage(null, 0L, null); + // persist msg to local tmp file. + pair.getRight().write(payloadData); + } + } else { + // non-txn test + payloadData = incrementedNumberMessageFormatter.formatMessage(); + } } else { payloadData = payloadBytes; } + TypedMessageBuilder messageBuilder; if (arguments.isEnableTransaction) { if (arguments.numMessagesPerTransaction > 0) { @@ -694,6 +971,15 @@ private static void runProducer(int producerId, if (arguments.exitOnFailure) { PerfClientUtils.exit(1); } + if (arguments.resendOnFailure && !arguments.isEnableTxnTest + && (messageFormatter instanceof IncrementedNumberMessageFormatter)) { + try { + ((IncrementedNumberMessageFormatter) messageFormatter).msgToResend.put(payloadData); + totalSent.decrement(); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } return null; }); if (arguments.isEnableTransaction @@ -707,11 +993,35 @@ private static void runProducer(int producerId, } totalEndTxnOpSuccessNum.increment(); numTxnOpSuccess.increment(); + if (arguments.isEnableTxnTest) { + handleTxnOnCommitted(transaction); + } }) .exceptionally(exception -> { - log.error("Commit transaction failed with exception : ", - exception); + log.error("Commit transaction:{} failed with exception : ", + transaction.getTxnID(), exception); totalEndTxnOpFailNum.increment(); + + if (arguments.isEnableTxnTest) { + int retryTimes = arguments.retryTimes; + if (isTimeOutException(exception)) { + for (int i = 0; i < retryTimes; i++) { + try { + transaction.commit().get(); + handleTxnOnCommitted(transaction); + break; + } catch (Exception e) { + if (!isTimeOutException(e)) { + break; + } + } + } + } + if (transaction.getState() != Transaction.State.COMMITTED) { + totalSent.add(-handleTxnOnAborted(transaction, messageFormatter, + arguments.baseDirToSaveResendTxnData)); + } + } return null; }); } else { @@ -735,6 +1045,12 @@ private static void runProducer(int producerId, .withTransactionTimeout(arguments.transactionTimeout, TimeUnit.SECONDS).build().get(); transactionAtomicReference.compareAndSet(transaction, newTransaction); + + if (arguments.isEnableTxnTest) { + File tmpFile = tmpFiles.take(); + openTmpFileStreamForNewTxn(tmpFile, newTransaction); + } + numMessageSend.set(0); numMsgPerTxnLimit.release(arguments.numMessagesPerTransaction); totalNumTxnOpenTxnSuccess.increment();