benchmark

benchmark copied to clipboard

The original code and fix are as follows.

  public CompletableFuture<BenchmarkProducer> createProducer(String topic) {
        if (config.producerType.equals(ClientType.KAFKA)) {
            final BenchmarkProducer producer =
                    new KafkaBenchmarkProducer(new KafkaProducer<>(producerProperties), topic);
            producers.add(producer);
            return CompletableFuture.completedFuture(producer);
        } else if (config.consumerType.equals(ClientType.PULSAR)) {
//This condition should be “config.producerType.equals(ClientType.PULSAR) ”
            return producerBuilder
                    .clone()
                    .topic(topic)
                    .createAsync()
                    .thenApply(PulsarBenchmarkProducer::new);
        } else {
            throw new IllegalArgumentException("producerType " + config.producerType + " is invalid");
        }
    }