I can help with some example source code. But as Tom says: do you want Direct messaging (best effort) or Guaranteed messaging (fully persistent, can't get lost)?

Hello Inder! So many questions, I will try to answer them all and hopefully not miss any. :-)

I also have to break this response into several sections because it's too long!!

First, some confusion around Topics vs. Topic Endpoints vs. Subscriptions in Solace. Endpoints are only when you need Persistence. For Direct messaging, consumer apps subscribe to topics (with wildcards?), and publishers publish on topics. Here's an awesome blog from Tom on Topics and Subscriptions: https://solace.com/blog/when-topics-arent-just-publish-subscribe/ and here's one that explains what Topic Endpoints are and when to use them: https://solace.com/blog/queues-vs-topic-endpoints/

Glad you figured out SdkPerf. We use it a lot for perf testing, it's super useful. And yeah, for Direct messaging, you just need to use the -ptl and -stl options, and -mt=direct. To improve throughput, make sure the network distance between your SdkPerf apps and the broker is minimized. I would think SdkPerf should be able to push out more than 3000 msg/s @ 300 bytes..?

You were asking why there are 2 subscriptions instead of 1 for your consumer? Every Solace client when it connects gets assigned a unique "inbox" topic subscription for receiving control messages from the broker. So a publish only app would have 1 subscription, and a consumer application with 1 added subscription would have 2. In the PubSub+ Manager GUI, you can click on the client name, and then click on the Subscriptions tab to see them.

Ok, so what kind of rates should you expect with one publisher and one consumer? Note that if you're using the Standard edition (free) Solace PubSub+ broker, it will be rate-limited to 10,000 msg/s. Otherwise, if using the Enterprise (paid), or Evaluation (90-day free: https://solace.com/downloads/#pubsub-enterprise-evaluation) version, or the hardware appliance, you should be able to get a LOT more... especially for Direct messaging. I'll show you what I got with my test pub and sub apps in a minute.

In your publisher, don't worry about the sendMultiple()... not really necessary. You're also using a TextMessage to send a byte array? Just use BytesMessages. And in the consumer, you always want to be in async receiving mode, where you register a callback Listener in the Consumer interface. Always the fastest and most efficient not to have to poll or do a receive() call.

And the consumer code:

/*
 * 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 com.solace.samples.aaron;

import java.io.IOException;

import com.solacesystems.jcsmp.BytesMessage;
import com.solacesystems.jcsmp.BytesXMLMessage;
import com.solacesystems.jcsmp.JCSMPException;
import com.solacesystems.jcsmp.JCSMPFactory;
import com.solacesystems.jcsmp.JCSMPGlobalProperties;
import com.solacesystems.jcsmp.JCSMPProperties;
import com.solacesystems.jcsmp.JCSMPSession;
import com.solacesystems.jcsmp.Topic;
import com.solacesystems.jcsmp.XMLMessageConsumer;
import com.solacesystems.jcsmp.XMLMessageListener;

public class FastDirectConsumer {

    private static volatile int msgCounter = 0;
    private static volatile boolean discardFlag = false;
    private static volatile boolean shutdown = false;

    public static void main(String... args) throws JCSMPException, IOException, InterruptedException {

        // Check command line arguments
        if (args.length < 2 || args[1].split("@").length != 2) {
            System.out.println("Usage: FastDirectConsumer <host:port> <client-username@message-vpn> [client-password]");
            System.out.println();
            System.exit(-1);
        }
        if (args[1].split("@")[0].isEmpty()) {
            System.out.println("No client-username entered");
            System.out.println();
            System.exit(-1);
        }
        if (args[1].split("@")[1].isEmpty()) {
            System.out.println("No message-vpn entered");
            System.out.println();
            System.exit(-1);
        }

        System.out.println("FastDirectConsumer initializing...");
        final JCSMPProperties properties = new JCSMPProperties();
        properties.setProperty(JCSMPProperties.HOST, args[0]);     // host:port
        properties.setProperty(JCSMPProperties.USERNAME, args[1].split("@")[0]); // client-username
        properties.setProperty(JCSMPProperties.VPN_NAME,  args[1].split("@")[1]); // message-vpn
        if (args.length > 2) {
            properties.setProperty(JCSMPProperties.PASSWORD, args[2]); // client-password
        }
        //properties.setProperty(JCSMPProperties.MESSAGE_CALLBACK_ON_REACTOR,true);  // use only 1 thread instead of 2, but hurts throughput
        {  // this next block is probably unnecessary until receiving > 100k msg/s, so delete if going slower than that
            JCSMPGlobalProperties gp = new JCSMPGlobalProperties();
            gp.setConsumerDefaultFlowCongestionLimit(20000);  // override default (5000)
            JCSMPFactory.onlyInstance().setGlobalProperties(gp);
        }
        final JCSMPSession session = JCSMPFactory.onlyInstance().createSession(properties);

        /** Anonymous inner-class for MessageListener
         *  This demonstrates the async threaded message callback */
        final XMLMessageConsumer cons = session.getMessageConsumer(new XMLMessageListener() {
            @Override
            public void onReceive(BytesXMLMessage msg) {
                // do you want to do anything with this message?
                msgCounter++;
                if (msg.getDiscardIndication()) {  // lost any messages?
                    // If the consumer is being over-driven (i.e. publish rates too high), the broker might discard some messages for this consumer
                    // check this flag to know if that's happened
                    // to avoid discards:
                    //  a) reduce publish rate
                    //  b) increase size of consumer's D-1 egress buffers (check client-profile)
                    //  c) use multiple-threads or shared subscriptions for parallel processing
                    discardFlag = true;  // set my own flag
                }
                // this next block is just to have a non-trivial onReceive() callback... let's do a bit of work
                boolean doSomeSillyVerifyingOfData = true;  // set this to false if publisher using single/static topic
                if (doSomeSillyVerifyingOfData) {
                    // as set in the publisher code, the payload should be filled with the same character as the last letter of the topic
                    BytesMessage message = (BytesMessage)msg;
                    if (message.getAttachmentContentLength() > 0) {
                        byte[] payload = message.getData();
                        char payloadChar = (char)payload[0];
                        char lastTopicChar = message.getDestination().getName().charAt(message.getDestination().getName().length()-1);
                        if (payloadChar != lastTopicChar) {
                            System.out.println("*** Topic vs. Payload discrepancy *** : didn't match! oh well!");
                            doSomeSillyVerifyingOfData = false;  // don't do any further testing
                        }
                    }
                }
            }

            @Override
            public void onException(JCSMPException e) {
                // uh oh!
                System.err.println("We've had an exception thrown to the Consumer's onException()");
                e.printStackTrace();
            }
        });

        session.connect();
        final Topic topic = JCSMPFactory.onlyInstance().createTopic("aaron/>");
        session.addSubscription(topic);
        cons.start();

        Runnable statsThread = new Runnable() {
            @Override
            public void run() {
                try {
                    while (!shutdown) {
                        Thread.sleep(1000);  // wait 1 second
                        System.out.printf("Msgs/s: %,d%n",msgCounter);
                        // kinda hacky way of doing message rates, but doesn't matter too much
                        msgCounter = 0;
                        if (discardFlag) {
                            System.out.println("*** Egress discard detected *** : consumer unable to keep up with full message rate");
                            discardFlag = false;  // only show the error once per second
                        }
                    }
                } catch (InterruptedException e) {
                    System.out.println("I was awoken while waiting");
                }
            }
        };
        Thread t = new Thread(statsThread,"Stats Thread");
        t.setDaemon(true);
        t.start();

        System.out.println("Connected, and running. Press [ENTER] to quit.");
        System.in.read();  // wait for user to end
        System.out.println("Quitting in 1 second.");
        shutdown = true;
        session.removeSubscription(topic);
        Thread.sleep(1000);
        // Close consumer
        cons.close();
        session.closeSession();
    }
}

Hey guys... thanks for the additional comments. And glad you liked my answer + code!

Also happy that you're upgrading your network to 1Gbps... that will help. While the bandwidth is not huge, there's also the serialization speed... 1Gbps puts data out onto the network 10x faster, so latencies will be lower.

Anyhow, ok... troubleshooting: first, always try to simplify things... run your publisher and subscriber SdkPerfs separately... that will (should?) help isolate who is having the problem.

If you only publish, does the same performance degradation happen? If publishing at a lower rate, does the same thing happen?

If it's the publisher having problems, does adding more subscribers (i.e. increasing the "fanout") change the behaviour? I'd hope not.

And you're sure you're publishing Direct, right? If publishing Guaranteed, and the disk on the broker was very slow, I could see that maybe eventually impacting the publishing performance..?

If your network is lossy and you're getting TCP transmissions, that could explain some drop in performance, but going from 4k down to 500 sounds pretty bad. The logs on the broker won't tell you much... they report connections/disconnections, stuff like that. You need to look at stats. Does ifotp show your TCP window information? Or retransmissions?

If you go into the the PubSub+ Manager (web GUI) then Client Connections, Solace Clients, click on them, and then under "Connections" you can see TCP errors. Let me know if there's anything report ed there.

That's a good start for debugging. Hope that helps! Let me know.