QPID-3936: Produce aggregated batchSize and bring acknowledgeMode through to CSV file

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1330959 13f79535-47bb-0310-9956-ffa450edef68

8 files changed, 89 insertions, 37 deletions

diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ConsumerParticipant.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ConsumerParticipant.java
index 7d68a8d84c..1b5e8276c2 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ConsumerParticipant.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ConsumerParticipant.java
@@ -65,6 +65,7 @@ public class ConsumerParticipant implements Participant
     public ParticipantResult doIt(String registeredClientName) throws Exception
     {
         final Date start = new Date();
+        final int acknowledgeMode = _jmsDelegate.getAcknowledgeMode(_command.getSessionName());
 
         if (_command.getMaximumDuration() == 0 && _command.getNumberOfMessages() == 0)
         {
@@ -100,11 +101,11 @@ public class ConsumerParticipant implements Participant
                 getName(),
                 registeredClientName,
                 _command,
+                acknowledgeMode,
                 numberOfMessagesSent,
                 payloadSize,
                 totalPayloadSize,
-                start,
-                end);
+                start, end);
 
         return result;
     }
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantResultFactory.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantResultFactory.java
index 2e4cda1460..7f6b96b87c 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantResultFactory.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantResultFactory.java
@@ -29,11 +29,11 @@ import org.apache.qpid.disttest.message.ProducerParticipantResult;
 
 public class ParticipantResultFactory
 {
-    public ConsumerParticipantResult createForConsumer(String participantName, String clientRegisteredName, CreateConsumerCommand command, int numberOfMessagesReceived, int payloadSize, long totalPayloadReceived, Date start, Date end)
+    public ConsumerParticipantResult createForConsumer(String participantName, String clientRegisteredName, CreateConsumerCommand command, int acknowledgeMode, int numberOfMessagesReceived, int payloadSize, long totalPayloadReceived, Date start, Date end)
     {
         ConsumerParticipantResult consumerParticipantResult = new ConsumerParticipantResult();
 
-        setTestProperties(consumerParticipantResult, command, participantName, clientRegisteredName);
+        setTestProperties(consumerParticipantResult, command, participantName, clientRegisteredName, acknowledgeMode);
         setTestResultProperties(consumerParticipantResult, numberOfMessagesReceived, payloadSize, totalPayloadReceived, start, end);
 
         consumerParticipantResult.setTopic(command.isTopic());
@@ -48,7 +48,7 @@ public class ParticipantResultFactory
         return consumerParticipantResult;
     }
 
-    public ProducerParticipantResult createForProducer(String participantName, String clientRegisteredName, CreateProducerCommand command, int numberOfMessagesSent, int payloadSize, long totalPayloadSent, Date start, Date end)
+    public ProducerParticipantResult createForProducer(String participantName, String clientRegisteredName, CreateProducerCommand command, int acknowledgeMode, int numberOfMessagesSent, int payloadSize, long totalPayloadSent, Date start, Date end)
     {
         final ProducerParticipantResult producerParticipantResult = new ProducerParticipantResult();
 
@@ -61,7 +61,7 @@ public class ParticipantResultFactory
         producerParticipantResult.setTotalNumberOfProducers(1);
 
 
-        setTestProperties(producerParticipantResult, command, participantName, clientRegisteredName);
+        setTestProperties(producerParticipantResult, command, participantName, clientRegisteredName, acknowledgeMode);
 
         setTestResultProperties(producerParticipantResult, numberOfMessagesSent, payloadSize, totalPayloadSent, start, end);
 
@@ -77,12 +77,13 @@ public class ParticipantResultFactory
         participantResult.setEndDate(end);
     }
 
-    private void setTestProperties(final ParticipantResult participantResult, CreateParticpantCommand command, String participantName, String clientRegisteredName)
+    private void setTestProperties(final ParticipantResult participantResult, CreateParticpantCommand command, String participantName, String clientRegisteredName, int acknowledgeMode)
     {
         participantResult.setParticipantName(participantName);
         participantResult.setRegisteredClientName(clientRegisteredName);
         participantResult.setBatchSize(command.getBatchSize());
         participantResult.setMaximumDuration(command.getMaximumDuration());
+        participantResult.setAcknowledgeMode(acknowledgeMode);
 
     }
 
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ProducerParticipant.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ProducerParticipant.java
index d40a62f947..bcad81b4aa 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ProducerParticipant.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ProducerParticipant.java
@@ -57,6 +57,8 @@ public class ProducerParticipant implements Participant
             throw new DistributedTestException("number of messages and duration cannot both be zero");
         }
 
+        int acknowledgeMode = _jmsDelegate.getAcknowledgeMode(_command.getSessionName());
+
         long expectedDuration = _command.getMaximumDuration() - _command.getStartDelay();
 
         doSleepForStartDelay();
@@ -126,9 +128,9 @@ public class ProducerParticipant implements Participant
                 getName(),
                 registeredClientName,
                 _command,
+                acknowledgeMode,
                 numberOfMessagesSent,
-                payloadSize,
-                totalPayloadSizeOfAllMessagesSent, start, end);
+                payloadSize, totalPayloadSizeOfAllMessagesSent, start, end);
     }
 
     private int getPayloadSizeForResultIfConstantOrZeroOtherwise(NavigableSet<Integer> allPayloadSizes)
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/jms/ClientJmsDelegate.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/jms/ClientJmsDelegate.java
index 282b933607..d68fc86a0e 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/jms/ClientJmsDelegate.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/jms/ClientJmsDelegate.java
@@ -360,6 +360,20 @@ public class ClientJmsDelegate
         }
     }
 
+    public int getAcknowledgeMode(final String sessionName)
+    {
+        try
+        {
+            final Session session = _testSessions.get(sessionName);
+            return session.getAcknowledgeMode();
+        }
+        catch (final JMSException jmse)
+        {
+            throw new DistributedTestException("Unable to determine acknowledgement mode for session: " +
+                            sessionName, jmse);
+        }
+    }
+
     public Message sendNextMessage(final CreateProducerCommand command)
     {
         Message sentMessage = null;
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/message/ParticipantAttribute.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/message/ParticipantAttribute.java
index 96bc4183de..ccc7c0d9fb 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/message/ParticipantAttribute.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/message/ParticipantAttribute.java
@@ -37,6 +37,7 @@ public enum ParticipantAttribute
     PAYLOAD_SIZE("payloadSizeB"),
     PRIORITY("priority"),
     TIME_TO_LIVE("timeToLiveMs"),
+    ACKNOWLEDGE_MODE("acknowledgeMode"),
     DELIVERY_MODE("deliveryMode"),
     BATCH_SIZE("batchSize"),
     MAXIMUM_DURATION("maximumDurationMs"),
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/message/ParticipantResult.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/message/ParticipantResult.java
index 576babf7a8..a6d3d91bae 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/message/ParticipantResult.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/message/ParticipantResult.java
@@ -36,11 +36,12 @@ public class ParticipantResult extends Response
 {
     private String _testName;
     private String _participantName;
+    private int _iterationNumber;
+
     private long _startInMillis;
     private long _endInMillis;
     private int _batchSize;
     private long _maximumDuration;
-    private int _iterationNumber;
 
     private String _configuredClientName;
 
@@ -52,6 +53,9 @@ public class ParticipantResult extends Response
     private int _totalNumberOfConsumers;
     private int _totalNumberOfProducers;
 
+    // As Session.SESSION_TRANSACTED is 0, we use value -1 so we can distinguish the case where an aggregated result
+    // summarizes results from participants using different session acknowledge modes.
+    private int _acknowledgeMode = -1;
 
     public static final Comparator<? super ParticipantResult> PARTICIPANT_NAME_COMPARATOR = new Comparator<ParticipantResult>()
     {
@@ -254,4 +258,15 @@ public class ParticipantResult extends Response
         return _totalNumberOfProducers;
     }
 
+    @OutputAttribute(attribute=ParticipantAttribute.ACKNOWLEDGE_MODE)
+    public int getAcknowledgeMode()
+    {
+        return _acknowledgeMode;
+    }
+
+    public void setAcknowledgeMode(int acknowledgeMode)
+    {
+        _acknowledgeMode = acknowledgeMode;
+    }
+
 }
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/ParticipantResultAggregator.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/ParticipantResultAggregator.java
index 603f475a16..207d0131eb 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/ParticipantResultAggregator.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/ParticipantResultAggregator.java
@@ -40,6 +40,8 @@ public class ParticipantResultAggregator
 
     private NavigableSet<Integer> _encounteredPayloadSizes = new TreeSet<Integer>();
     private NavigableSet<Integer> _encounteredIterationNumbers = new TreeSet<Integer>();
+    private NavigableSet<Integer> _encounteredBatchSizes = new TreeSet<Integer>();
+    private NavigableSet<Integer> _encounteredAcknowledgeMode = new TreeSet<Integer>();
     private NavigableSet<String> _encountedTestNames = new TreeSet<String>();
 
     public ParticipantResultAggregator(Class<? extends ParticipantResult> targetClass, String aggregateResultName)
@@ -90,6 +92,8 @@ public class ParticipantResultAggregator
         }
         _encounteredPayloadSizes.add(result.getPayloadSize());
         _encounteredIterationNumbers.add(result.getIterationNumber());
+        _encounteredBatchSizes.add(result.getBatchSize());
+        _encounteredAcknowledgeMode.add(result.getAcknowledgeMode());
     }
 
     private void setComputedVariableAttributes(ParticipantResult aggregatedResult)
@@ -117,6 +121,14 @@ public class ParticipantResultAggregator
         {
             aggregatedResult.setTestName(_encountedTestNames.first());
         }
+        if (_encounteredBatchSizes.size() == 1)
+        {
+            aggregatedResult.setBatchSize(_encounteredBatchSizes.first());
+        }
+        if (_encounteredAcknowledgeMode.size() == 1)
+        {
+            aggregatedResult.setAcknowledgeMode(_encounteredAcknowledgeMode.first());
+        }
     }
 
     private double calculateThroughputInKiloBytesPerSecond()
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/TestResultAggregator.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/TestResultAggregator.java
index ac0e146635..5934e0e997 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/TestResultAggregator.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/TestResultAggregator.java
@@ -31,36 +31,52 @@ public class TestResultAggregator
 
     public AggregatedTestResult aggregateTestResult(ITestResult originalTestResult)
     {
-        AggregatedTestResult newTestResult = new AggregatedTestResult(originalTestResult);
+        ParticipantResultAggregator consumerResultAggregator = new ParticipantResultAggregator(ConsumerParticipantResult.class,
+                                                                                               ALL_CONSUMER_PARTICIPANTS_NAME);
+        ParticipantResultAggregator producerResultAggregator = new ParticipantResultAggregator(ProducerParticipantResult.class,
+                                                                                               ALL_PRODUCER_PARTICIPANTS_NAME);
+        ParticipantResultAggregator aggregatedResultsAggregator = new ParticipantResultAggregator(ParticipantResult.class,
+                                                                                               ALL_PARTICIPANTS_NAME);
 
-        ParticipantResultAggregator consumerParticipantResultAggregator = new ParticipantResultAggregator(ConsumerParticipantResult.class,
-                                                                                                          ALL_CONSUMER_PARTICIPANTS_NAME);
-        ParticipantResultAggregator producerParticipantResultAggregator = new ParticipantResultAggregator(ProducerParticipantResult.class,
-                                                                                                          ALL_PRODUCER_PARTICIPANTS_NAME);
+        boolean hasError = aggregateOriginalResults(originalTestResult,
+                                     consumerResultAggregator,
+                                     producerResultAggregator);
 
-        boolean hasError = aggregate(originalTestResult,
-                                     consumerParticipantResultAggregator,
-                                     producerParticipantResultAggregator);
+        ParticipantResult aggregatedProducerResult = producerResultAggregator.getAggregatedResult();
+        ParticipantResult aggregaredConsumerResult = consumerResultAggregator.getAggregatedResult();
 
-        ParticipantResult aggregatedProducerResult = producerParticipantResultAggregator.getAggregatedResult();
-        newTestResult.setAllProducerParticipantResult(aggregatedProducerResult);
+        ParticipantResult aggregatedAllResult = aggregateAggregatedResults(
+                aggregatedResultsAggregator, aggregatedProducerResult,
+                aggregaredConsumerResult);
 
-        ParticipantResult aggregaredConsumerResult = consumerParticipantResultAggregator.getAggregatedResult();
-        newTestResult.setAllConsumerParticipantResult(aggregaredConsumerResult);
-
-        ParticipantResult aggregatedAllResult = buildAllResultFromOtherAggregatedResults(
+        applyNonAggregateablesToAll(aggregatedAllResult,
                 aggregatedProducerResult, aggregaredConsumerResult);
 
+        AggregatedTestResult newTestResult = new AggregatedTestResult(originalTestResult);
+        newTestResult.setAllProducerParticipantResult(aggregatedProducerResult);
+        newTestResult.setAllConsumerParticipantResult(aggregaredConsumerResult);
+        newTestResult.setAllParticipantResult(aggregatedAllResult);
+
         if (hasError)
         {
             aggregatedAllResult.setErrorMessage(TestResultAggregator.AGGREGATED_ERROR_MESSAGE);
         }
-        newTestResult.setAllParticipantResult(aggregatedAllResult);
 
         return newTestResult;
     }
 
-    private boolean aggregate(ITestResult originalTestResult,
+    private ParticipantResult aggregateAggregatedResults(
+            ParticipantResultAggregator aggregatedResultsAggregator,
+            ParticipantResult aggregatedProducerResult,
+            ParticipantResult aggregaredConsumerResult)
+    {
+        aggregatedResultsAggregator.aggregate(aggregatedProducerResult);
+        aggregatedResultsAggregator.aggregate(aggregaredConsumerResult);
+        ParticipantResult aggregatedAllResult = aggregatedResultsAggregator.getAggregatedResult();
+        return aggregatedAllResult;
+    }
+
+    private boolean aggregateOriginalResults(ITestResult originalTestResult,
             ParticipantResultAggregator consumerParticipantResultAggregator,
             ParticipantResultAggregator producerParticipantResultAggregator)
     {
@@ -78,23 +94,13 @@ public class TestResultAggregator
         return hasError;
     }
 
-    private ParticipantResult buildAllResultFromOtherAggregatedResults(
-            ParticipantResult aggregatedProducerResult, ParticipantResult aggregatedConsumerResult)
+    private void applyNonAggregateablesToAll(ParticipantResult aggregatedAllResult, ParticipantResult aggregatedProducerResult, ParticipantResult aggregatedConsumerResult)
     {
-        ParticipantResult aggregatedAllResult = new ParticipantResult(ALL_PARTICIPANTS_NAME);
         aggregatedAllResult.setStartDate(aggregatedProducerResult.getStartDate());
-
         aggregatedAllResult.setEndDate(aggregatedConsumerResult.getEndDate());
 
-        aggregatedAllResult.setIterationNumber(aggregatedConsumerResult.getIterationNumber());
-        aggregatedAllResult.setTestName(aggregatedConsumerResult.getTestName());
         aggregatedAllResult.setNumberOfMessagesProcessed(aggregatedConsumerResult.getNumberOfMessagesProcessed());
-        aggregatedAllResult.setPayloadSize(aggregatedConsumerResult.getPayloadSize());
         aggregatedAllResult.setTotalPayloadProcessed(aggregatedConsumerResult.getTotalPayloadProcessed());
         aggregatedAllResult.setThroughput(aggregatedConsumerResult.getThroughput());
-        aggregatedAllResult.setTotalNumberOfProducers(aggregatedProducerResult.getTotalNumberOfProducers());
-        aggregatedAllResult.setTotalNumberOfConsumers(aggregatedConsumerResult.getTotalNumberOfConsumers());
-
-        return aggregatedAllResult;
     }
 }