QPID-5855: Fix to JAVA Client Can not recieve message with qpid ha cluster.

The original fix for this introduced a regression, running the qpid-txttest2
test against a cluster with the linear store failed. This fixes the fix.

- Run transaction commit logic when the commit completes. Report completion to the user only when
  all prior commands have completed (sync point)
- Fix missing initializer in client/amqp0_10/SessionImpl.cpp for transaction committing flag.
- Remove annoying log messages from IdSetter.h
- Skip transactional messages in prepare, don't wait till commit.
- Added fetch-timeout option to qpid-txtest2

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

1 files changed, 4 insertions, 2 deletions

diff --git a/qpid/cpp/src/tests/qpid-txtest2.cpp b/qpid/cpp/src/tests/qpid-txtest2.cpp
index 6e2f81726e..5e1dd492da 100644
--- a/qpid/cpp/src/tests/qpid-txtest2.cpp
+++ b/qpid/cpp/src/tests/qpid-txtest2.cpp
@@ -63,11 +63,12 @@ struct Options : public qpid::Options {
     qpid::log::Options log;
     uint port;
     bool quiet;
+    double fetchTimeout;
 
     Options() : help(false), init(true), transfer(true), check(true),
                 size(256), durable(true), queues(2),
                 base("tx"), msgsPerTx(1), txCount(5), totalMsgCount(10),
-                capacity(1000), url("localhost"), port(0), quiet(false)
+                capacity(1000), url("localhost"), port(0), quiet(false), fetchTimeout(5)
     {
         addOptions()
             ("init", qpid::optValue(init, "yes|no"), "Declare queues and populate one with the initial set of messages.")
@@ -85,6 +86,7 @@ struct Options : public qpid::Options {
             ("connection-options", qpid::optValue(connectionOptions, "OPTIONS"), "options for the connection")
             ("port,p", qpid::optValue(port, "PORT"), "(for test compatibility only, use broker option instead)")
             ("quiet", qpid::optValue(quiet), "reduce output from test")
+            ("fetch-timeout", qpid::optValue(fetchTimeout, "SECONDS"), "Timeout for transactional fetch")
             ("help", qpid::optValue(help), "print this usage statement");
         add(log);
     }
@@ -199,7 +201,7 @@ struct Transfer : public TransactionalClient, public Runnable
                 id << source << ">" << target << ":" << t+1;
                 try {
                     for (uint m = 0; m < opts.msgsPerTx; m++) {
-                        Message msg = receiver.fetch(Duration::SECOND*30);
+                        Message msg = receiver.fetch(Duration::SECOND*opts.fetchTimeout);
                         if (msg.getContentSize() != opts.size) {
                             std::ostringstream oss;
                             oss << "Message size incorrect: size=" << msg.getContentSize() << "; expected " << opts.size;