diff --git a/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/Interaction.java b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/Interaction.java
index 98ffe45..ab2c959 100644
--- a/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/Interaction.java
+++ b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/Interaction.java
@@ -20,12 +20,6 @@
 
 package org.apache.qpid.tests.protocol.v1_0;
 
-import static org.hamcrest.CoreMatchers.instanceOf;
-import static org.hamcrest.CoreMatchers.notNullValue;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.is;
-
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -117,6 +111,7 @@
     private Map<String, Object> _latestDeliveryApplicationProperties;
     private Map<Class, FrameBody> _latestResponses = new HashMap<>();
     private AtomicLong _receivedDeliveryCount = new AtomicLong();
+    private AtomicLong _coordinatorCredits = new AtomicLong();
 
     Interaction(final FrameTransport frameTransport)
     {
@@ -955,6 +950,12 @@
 
     public Interaction txnAttachCoordinatorLink(InteractionTransactionalState transactionalState) throws Exception
     {
+        return txnAttachCoordinatorLink(transactionalState, Accepted.ACCEPTED_SYMBOL, Rejected.REJECTED_SYMBOL);
+    }
+
+    public Interaction txnAttachCoordinatorLink(final InteractionTransactionalState transactionalState,
+                                                final Symbol... outcomes) throws Exception
+    {
         Attach attach = new Attach();
         attach.setName("testTransactionCoordinator-" + transactionalState.getHandle());
         attach.setHandle(transactionalState.getHandle());
@@ -963,78 +964,99 @@
         attach.setRole(Role.SENDER);
         Source source = new Source();
         attach.setSource(source);
-        source.setOutcomes(Accepted.ACCEPTED_SYMBOL, Rejected.REJECTED_SYMBOL);
+        source.setOutcomes(outcomes);
         sendPerformativeAndChainFuture(attach, _sessionChannel);
         consumeResponse(Attach.class);
-        consumeResponse(Flow.class);
+        final Flow flow = consumeResponse(Flow.class).getLatestResponse(Flow.class);
+        _coordinatorCredits.set(flow.getLinkCredit().longValue());
         return this;
     }
 
     public Interaction txnDeclare(final InteractionTransactionalState txnState) throws Exception
     {
-        Transfer transfer = createTransactionTransfer(txnState.getHandle());
-        transferPayload(transfer, new Declare());
-        sendPerformativeAndChainFuture(transfer, _sessionChannel);
-        consumeResponse(Disposition.class);
-        Disposition declareTransactionDisposition = getLatestResponse(Disposition.class);
-        assertThat(declareTransactionDisposition.getSettled(), is(equalTo(true)));
-        assertThat(declareTransactionDisposition.getState(), is(instanceOf(Declared.class)));
-        Binary transactionId = ((Declared) declareTransactionDisposition.getState()).getTxnId();
-        assertThat(transactionId, is(notNullValue()));
-        consumeResponse(Flow.class);
+        sendPayloadToCoordinator(new Declare(), txnState.getHandle());
+        final DeliveryState state = handleCoordinatorResponse();
+        txnState.setDeliveryState(state);
+        final Binary transactionId = ((Declared) state).getTxnId();
         txnState.setLastTransactionId(transactionId);
         return this;
     }
 
-    public Interaction discharge(final InteractionTransactionalState txnState, final boolean failed) throws Exception
+    public Interaction txnSendDischarge(final InteractionTransactionalState txnState, final boolean failed)
+            throws Exception
     {
         final Discharge discharge = new Discharge();
         discharge.setTxnId(txnState.getCurrentTransactionId());
         discharge.setFail(failed);
-
-        Transfer transfer = createTransactionTransfer(txnState.getHandle());
-        transferPayload(transfer, discharge);
-        sendPerformativeAndChainFuture(transfer, _sessionChannel);
+        sendPayloadToCoordinator(discharge, txnState.getHandle());
         return this;
     }
 
     public Interaction txnDischarge(final InteractionTransactionalState txnState, boolean failed) throws Exception
     {
-        discharge(txnState, failed);
-
-        Disposition declareTransactionDisposition = null;
-        Flow coordinatorFlow = null;
-        do
-        {
-            consumeResponse(Disposition.class, Flow.class);
-            Response<?> response = getLatestResponse();
-            if (response.getBody() instanceof Disposition)
-            {
-                declareTransactionDisposition = (Disposition) response.getBody();
-            }
-            if (response.getBody() instanceof Flow)
-            {
-                final Flow flowResponse = (Flow) response.getBody();
-                if (flowResponse.getHandle().equals(txnState.getHandle()))
-                {
-                    coordinatorFlow = flowResponse;
-                }
-            }
-        } while(declareTransactionDisposition == null || coordinatorFlow == null);
-
-        assertThat(declareTransactionDisposition.getSettled(), is(equalTo(true)));
-        assertThat(declareTransactionDisposition.getState(), is(instanceOf(Accepted.class)));
-
+        txnSendDischarge(txnState, failed);
+        final DeliveryState state = handleCoordinatorResponse();
+        txnState.setDeliveryState(state);
         txnState.setLastTransactionId(null);
         return this;
     }
 
+    private void sendPayloadToCoordinator(final Object payload, final UnsignedInteger handle)
+            throws Exception
+    {
+        final Transfer transfer = createTransactionTransfer(handle);
+        transferPayload(transfer, payload);
+        sendPerformativeAndChainFuture(transfer, _sessionChannel);
+    }
+
+    private DeliveryState handleCoordinatorResponse() throws Exception
+    {
+        final Set<Class<?>> expected = new HashSet<>(Collections.singletonList(Disposition.class));
+
+        if (_coordinatorCredits.decrementAndGet() == 0)
+        {
+            expected.add(Flow.class);
+        }
+
+        final Map<Class<?>, ?> responses = consumeResponses(expected);
+
+        final Disposition disposition = (Disposition) responses.get(Disposition.class);
+        if (expected.contains(Flow.class))
+        {
+            Flow flow = (Flow) responses.get(Flow.class);
+            _coordinatorCredits.set(flow.getLinkCredit().longValue());
+        }
+        if (!Boolean.TRUE.equals(disposition.getSettled()))
+        {
+            throw new IllegalStateException("Coordinator disposition is not settled");
+        }
+        return disposition.getState();
+    }
+
+    private Map<Class<?>, ?> consumeResponses(final Set<Class<?>> responseTypes)
+            throws Exception
+    {
+        Map<Class<?>, Object> results = new HashMap<>();
+        do
+        {
+            Response<?> response = consumeResponse(responseTypes).getLatestResponse();
+            if (response != null && response.getBody() instanceof FrameBody)
+            {
+                Class<?> bodyClass = response.getBody().getClass();
+                results.put(bodyClass, response.getBody());
+            }
+        }
+        while (!results.keySet().containsAll(responseTypes));
+        return results;
+    }
+
     private Transfer createTransactionTransfer(final UnsignedInteger handle)
     {
         Transfer transfer = new Transfer();
         transfer.setHandle(handle);
         transfer.setDeliveryId(getNextDeliveryId());
-        transfer.setDeliveryTag(new Binary(("transaction-" + transfer.getDeliveryId()).getBytes(StandardCharsets.UTF_8)));
+        transfer.setDeliveryTag(new Binary(("transaction-"
+                                            + transfer.getDeliveryId()).getBytes(StandardCharsets.UTF_8)));
         return transfer;
     }
 
diff --git a/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/InteractionTransactionalState.java b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/InteractionTransactionalState.java
index 061be92..b0832cc 100644
--- a/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/InteractionTransactionalState.java
+++ b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/InteractionTransactionalState.java
@@ -21,12 +21,14 @@
 package org.apache.qpid.tests.protocol.v1_0;
 
 import org.apache.qpid.server.protocol.v1_0.type.Binary;
+import org.apache.qpid.server.protocol.v1_0.type.DeliveryState;
 import org.apache.qpid.server.protocol.v1_0.type.UnsignedInteger;
 
 public class InteractionTransactionalState
 {
     private final UnsignedInteger _handle;
     private Binary _lastTransactionId;
+    private DeliveryState _deliveryState;
 
     public InteractionTransactionalState(final UnsignedInteger handle)
     {
@@ -47,4 +49,14 @@
     {
         return _lastTransactionId;
     }
+
+    public DeliveryState getDeliveryState()
+    {
+        return _deliveryState;
+    }
+
+    public void setDeliveryState(final DeliveryState deliveryState)
+    {
+        _deliveryState = deliveryState;
+    }
 }
diff --git a/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/Utils.java b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/Utils.java
index e846b5f..2ac50f4 100644
--- a/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/Utils.java
+++ b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/Utils.java
@@ -20,6 +20,7 @@
 
 package org.apache.qpid.tests.protocol.v1_0;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.is;
 import static org.junit.Assume.assumeThat;
@@ -28,6 +29,7 @@
 import java.util.stream.IntStream;
 
 import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.protocol.v1_0.type.Binary;
 import org.apache.qpid.server.protocol.v1_0.type.UnsignedInteger;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Accepted;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Header;
@@ -37,6 +39,7 @@
 import org.apache.qpid.server.protocol.v1_0.type.transport.Flow;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Open;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Role;
+import org.apache.qpid.server.protocol.v1_0.type.transport.SenderSettleMode;
 import org.apache.qpid.tests.utils.BrokerAdmin;
 
 public class Utils
@@ -83,7 +86,7 @@
                        .attachSourceAddress(queueName)
                        .attach().consumeResponse()
                        .flowIncomingWindow(UnsignedInteger.ONE)
-                       .flowNextIncomingId(UnsignedInteger.ZERO)
+                       .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
                        .flowOutgoingWindow(UnsignedInteger.ZERO)
                        .flowNextOutgoingId(UnsignedInteger.ZERO)
                        .flowLinkCredit(UnsignedInteger.ONE)
@@ -157,6 +160,7 @@
                                              .begin().consumeResponse(Begin.class)
                                              .attachRole(Role.SENDER)
                                              .attachTargetAddress(BrokerAdmin.TEST_QUEUE_NAME)
+                                             .attachSndSettleMode(SenderSettleMode.SETTLED)
                                              .attach().consumeResponse(Attach.class)
                                              .consumeResponse(Flow.class)
                                              .getLatestResponse(Flow.class);
@@ -166,14 +170,21 @@
                                          message.length),
                            flow.getLinkCredit().intValue(),
                            is(greaterThan(message.length)));
+
+                int tag = 0;
                 for (String payload : message)
                 {
                     interaction.transferPayloadData(payload)
                                .transferSettled(true)
+                               .transferDeliveryId()
+                               .transferDeliveryTag(new Binary(String.valueOf(tag).getBytes(UTF_8)))
                                .transfer()
                                .sync();
+                    tag++;
                 }
+                interaction.doCloseConnection();
             }
         }
     }
+
 }
diff --git a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/DecodeErrorTest.java b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/DecodeErrorTest.java
index 1ca4419..b079122 100644
--- a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/DecodeErrorTest.java
+++ b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/DecodeErrorTest.java
@@ -24,14 +24,18 @@
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.notNullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.is;
 import static org.junit.Assert.fail;
+import static org.junit.Assume.assumeThat;
 
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 
+import org.hamcrest.Matchers;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -39,6 +43,8 @@
 import org.apache.qpid.server.protocol.v1_0.codec.StringWriter;
 import org.apache.qpid.server.protocol.v1_0.type.Symbol;
 import org.apache.qpid.server.protocol.v1_0.type.UnsignedInteger;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.AmqpValue;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.AmqpValueSection;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.DeliveryAnnotations;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.DeliveryAnnotationsSection;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Header;
@@ -53,11 +59,13 @@
 import org.apache.qpid.server.protocol.v1_0.type.transport.Error;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Flow;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Open;
+import org.apache.qpid.server.protocol.v1_0.type.transport.ReceiverSettleMode;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Role;
 import org.apache.qpid.tests.protocol.Response;
 import org.apache.qpid.tests.protocol.SpecificationTest;
 import org.apache.qpid.tests.utils.BrokerAdmin;
 import org.apache.qpid.tests.utils.BrokerAdminUsingTestBase;
+import org.apache.qpid.tests.utils.BrokerSpecific;
 
 public class DecodeErrorTest extends BrokerAdminUsingTestBase
 {
@@ -74,44 +82,46 @@
     @SpecificationTest(section = "3.2",
             description = "Altogether a message consists of the following sections: Zero or one header,"
                           + " Zero or one delivery-annotations, [...]")
-    public void illegalMessageFormatPayload() throws Exception
+    @BrokerSpecific(kind = BrokerAdmin.KIND_BROKER_J)
+    public void illegalMessage() throws Exception
     {
         try (FrameTransport transport = new FrameTransport(_brokerAddress).connect())
         {
+            final Interaction interaction = transport.newInteraction();
+            final Attach attach = interaction.negotiateProtocol()
+                                             .consumeResponse()
+                                             .open()
+                                             .consumeResponse(Open.class)
+                                             .begin()
+                                             .consumeResponse(Begin.class)
+                                             .attachRole(Role.SENDER)
+                                             .attachTargetAddress(BrokerAdmin.TEST_QUEUE_NAME)
+                                             .attachRcvSettleMode(ReceiverSettleMode.SECOND)
+                                             .attach()
+                                             .consumeResponse(Attach.class)
+                                             .getLatestResponse(Attach.class);
+            assumeThat(attach.getRcvSettleMode(), is(equalTo(ReceiverSettleMode.SECOND)));
 
-            List<QpidByteBuffer> payloads = new ArrayList<>();
-            final HeaderSection headerSection = new Header().createEncodingRetainingSection();
-            payloads.add(headerSection.getEncodedForm());
-            headerSection.dispose();
-            final StringWriter stringWriter = new StringWriter("string in between annotation sections");
-            QpidByteBuffer encodedString = QpidByteBuffer.allocate(stringWriter.getEncodedSize());
-            stringWriter.writeToBuffer(encodedString);
-            encodedString.flip();
-            payloads.add(encodedString);
-            final DeliveryAnnotationsSection
-                    deliveryAnnotationsSection =
-                    new DeliveryAnnotations(Collections.emptyMap()).createEncodingRetainingSection();
-            payloads.add(deliveryAnnotationsSection.getEncodedForm());
-            deliveryAnnotationsSection.dispose();
+            final Flow flow = interaction.consumeResponse(Flow.class).getLatestResponse(Flow.class);
+            assumeThat(flow.getLinkCredit(), is(greaterThan(UnsignedInteger.ZERO)));
 
-            final Detach detachResponse;
-            try (QpidByteBuffer combinedPayload = QpidByteBuffer.concatenate(payloads))
+            final List<QpidByteBuffer> payloads = buildInvalidMessage();
+            try
             {
-                detachResponse = transport.newInteraction()
-                                          .negotiateProtocol().consumeResponse()
-                                          .open().consumeResponse(Open.class)
-                                          .begin().consumeResponse(Begin.class)
-                                          .attachRole(Role.SENDER)
-                                          .attachTargetAddress(BrokerAdmin.TEST_QUEUE_NAME)
-                                          .attach().consumeResponse(Attach.class)
-                                          .consumeResponse(Flow.class)
-                                          .transferMessageFormat(UnsignedInteger.ZERO)
-                                          .transferPayload(combinedPayload)
-                                          .transfer()
-                                          .consumeResponse()
-                                          .getLatestResponse(Detach.class);
+                try (QpidByteBuffer combinedPayload = QpidByteBuffer.concatenate(payloads))
+                {
+                    interaction.transferMessageFormat(UnsignedInteger.ZERO)
+                               .transferPayload(combinedPayload)
+                               .transfer();
+                }
             }
-            payloads.forEach(QpidByteBuffer::dispose);
+            finally
+            {
+                payloads.forEach(QpidByteBuffer::dispose);
+            }
+
+            final Detach detachResponse = interaction.consumeResponse()
+                                                     .getLatestResponse(Detach.class);
             assertThat(detachResponse.getError(), is(notNullValue()));
             assertThat(detachResponse.getError().getCondition(), is(equalTo(DECODE_ERROR)));
         }
@@ -148,6 +158,10 @@
             {
                 error = ((Close) responseBody).getError();
             }
+            else if (responseBody instanceof Detach)
+            {
+                error = ((Detach) responseBody).getError();
+            }
             else
             {
                 fail(String.format("Expected response of either Detach, End, or Close. Got '%s'", responseBody));
@@ -200,4 +214,51 @@
             assertThat(error.getCondition(), is(equalTo(DECODE_ERROR)));
         }
     }
+
+    private List<QpidByteBuffer> buildInvalidMessage()
+    {
+        final List<QpidByteBuffer> payloads = new ArrayList<>();
+        final Header header = new Header();
+        header.setTtl(UnsignedInteger.valueOf(1000L));
+        final HeaderSection headerSection = header.createEncodingRetainingSection();
+        try
+        {
+            payloads.add(headerSection.getEncodedForm());
+        }
+        finally
+        {
+            headerSection.dispose();
+        }
+
+        final StringWriter stringWriter = new StringWriter("string in between annotation sections");
+        QpidByteBuffer encodedString = QpidByteBuffer.allocate(stringWriter.getEncodedSize());
+        stringWriter.writeToBuffer(encodedString);
+        encodedString.flip();
+        payloads.add(encodedString);
+
+        final Map<Symbol, Object> annoationMap = Collections.singletonMap(Symbol.valueOf("foo"), "bar");
+        final DeliveryAnnotations annotations = new DeliveryAnnotations(annoationMap);
+        final DeliveryAnnotationsSection deliveryAnnotationsSection = annotations.createEncodingRetainingSection();
+        try
+        {
+
+            payloads.add(deliveryAnnotationsSection.getEncodedForm());
+        }
+        finally
+        {
+            deliveryAnnotationsSection.dispose();
+        }
+
+        final AmqpValueSection payload = new AmqpValue(getTestName()).createEncodingRetainingSection();
+        try
+        {
+            payloads.add(payload.getEncodedForm());
+        }
+        finally
+        {
+            payload.dispose();
+        }
+        return payloads;
+    }
+
 }
diff --git a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/extensions/anonymousterminus/AnonymousTerminusTest.java b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/extensions/anonymousterminus/AnonymousTerminusTest.java
index 1878950..a7ef076 100644
--- a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/extensions/anonymousterminus/AnonymousTerminusTest.java
+++ b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/extensions/anonymousterminus/AnonymousTerminusTest.java
@@ -46,8 +46,6 @@
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Properties;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Rejected;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Target;
-import org.apache.qpid.server.protocol.v1_0.type.transaction.Coordinator;
-import org.apache.qpid.server.protocol.v1_0.type.transaction.Discharge;
 import org.apache.qpid.server.protocol.v1_0.type.transaction.TransactionError;
 import org.apache.qpid.server.protocol.v1_0.type.transaction.TransactionalState;
 import org.apache.qpid.server.protocol.v1_0.type.transport.AmqpError;
@@ -104,6 +102,7 @@
                        .attach().consumeResponse(Attach.class)
                        .consumeResponse(Flow.class)
 
+                       .transferDeliveryId()
                        .transferPayload(generateMessagePayloadToDestination(BrokerAdmin.TEST_QUEUE_NAME))
                        .transferSettled(Boolean.TRUE)
                        .transferDeliveryTag(_deliveryTag)
@@ -139,12 +138,13 @@
                        .attach().consumeResponse(Attach.class)
                        .consumeResponse(Flow.class)
 
+                       .transferDeliveryId()
                        .transferPayload(generateMessagePayloadToDestination("Unknown"))
                        .transferSettled(Boolean.TRUE)
                        .transferDeliveryTag(_deliveryTag)
                        .transfer();
 
-            Detach detach = interaction.consumeResponse().getLatestResponse(Detach.class);
+            Detach detach = interaction.consumeResponse(Detach.class).getLatestResponse(Detach.class);
             Error error = detach.getError();
             assertThat(error, is(notNullValue()));
             assertThat(error.getCondition(), is(equalTo(AmqpError.NOT_FOUND)));
@@ -179,6 +179,7 @@
                        .attach().consumeResponse(Attach.class)
                        .consumeResponse(Flow.class)
 
+                       .transferDeliveryId()
                        .transferPayload(generateMessagePayloadToDestination("Unknown"))
                        .transferDeliveryTag(_deliveryTag)
                        .transfer()
@@ -226,6 +227,7 @@
                        .attach().consumeResponse(Attach.class)
                        .consumeResponse(Flow.class)
 
+                       .transferDeliveryId()
                        .transferPayload(generateMessagePayloadToDestination("Unknown"))
                        .transferDeliveryTag(_deliveryTag)
                        .transfer();
@@ -262,6 +264,7 @@
                        .attach().consumeResponse(Attach.class)
                        .consumeResponse(Flow.class)
 
+                       .transferDeliveryId()
                        .transferHandle(linkHandle)
                        .transferPayload(generateMessagePayloadToDestination(BrokerAdmin.TEST_QUEUE_NAME))
                        .transferDeliveryTag(_deliveryTag)
@@ -271,6 +274,8 @@
 
                        .txnDischarge(txnState, false);
 
+            assertThat(txnState.getDeliveryState(), is(instanceOf(Accepted.class)));
+
             Object receivedMessage = Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME);
             assertThat(receivedMessage, is(equalTo(TEST_MESSAGE_CONTENT)));
         }
@@ -295,6 +300,7 @@
                        .attach().consumeResponse(Attach.class)
                        .consumeResponse(Flow.class)
 
+                       .transferDeliveryId()
                        .transferHandle(linkHandle)
                        .transferPayload(generateMessagePayloadToDestination(BrokerAdmin.TEST_QUEUE_NAME))
                        .transferDeliveryTag(_deliveryTag)
@@ -314,6 +320,8 @@
 
             interaction.txnDischarge(txnState, false);
 
+            assertThat(txnState.getDeliveryState(), is(instanceOf(Accepted.class)));
+
             Object receivedMessage = Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME);
             assertThat(receivedMessage, is(equalTo(TEST_MESSAGE_CONTENT)));
         }
@@ -339,6 +347,7 @@
                        .attach().consumeResponse(Attach.class)
                        .consumeResponse(Flow.class)
 
+                       .transferDeliveryId()
                        .transferHandle(linkHandle)
                        .transferPayload(generateMessagePayloadToDestination("Unknown"))
                        .transferDeliveryTag(_deliveryTag)
@@ -362,6 +371,8 @@
             assertThat(rejectedError.getInfo().get(DELIVERY_TAG), is(equalTo(_deliveryTag)));
 
             interaction.txnDischarge(txnState, false);
+
+            assertThat(txnState.getDeliveryState(), is(instanceOf(Accepted.class)));
         }
     }
 
@@ -385,6 +396,7 @@
                        .attach().consumeResponse(Attach.class)
                        .consumeResponse(Flow.class)
 
+                       .transferDeliveryId()
                        .transferHandle(linkHandle)
                        .transferPayload(generateMessagePayloadToDestination("Unknown"))
                        .transferDeliveryId(UnsignedInteger.valueOf(1))
@@ -400,23 +412,11 @@
             assertThat(senderLinkDetachError.getInfo(), is(notNullValue()));
             assertThat(senderLinkDetachError.getInfo().get(DELIVERY_TAG), is(equalTo(_deliveryTag)));
 
-            final Discharge discharge = new Discharge();
-            discharge.setTxnId(txnState.getCurrentTransactionId());
-            discharge.setFail(false);
+            interaction.txnDischarge(txnState, false);
 
-            interaction.transferHandle(txnState.getHandle())
-                       .transferSettled(Boolean.FALSE)
-                       .transferDeliveryId(UnsignedInteger.valueOf(2))
-                       .transferDeliveryTag(new Binary(("transaction-" + 2).getBytes(StandardCharsets.UTF_8)))
-                       .transferPayloadData(discharge).transfer();
-
-            Disposition dischargeTransactionDisposition =
-                    getDispositionForDeliveryId(interaction, UnsignedInteger.valueOf(2));
-
-            assertThat(dischargeTransactionDisposition.getSettled(), is(equalTo(true)));
-            assertThat(dischargeTransactionDisposition.getState(), is(instanceOf(Rejected.class)));
-
-            Rejected rejected = (Rejected) dischargeTransactionDisposition.getState();
+            DeliveryState txnDischargeDeliveryState = txnState.getDeliveryState();
+            assertThat(txnDischargeDeliveryState, is(instanceOf(Rejected.class)));
+            Rejected rejected = (Rejected) txnDischargeDeliveryState;
             Error error = rejected.getError();
 
             assertThat(error, is(notNullValue()));
@@ -468,30 +468,20 @@
                        .attach().consumeResponse(Attach.class)
                        .consumeResponse(Flow.class)
 
+                       .transferDeliveryId()
                        .transferHandle(linkHandle)
                        .transferPayload(generateMessagePayloadToDestination("Unknown"))
                        .transferDeliveryTag(_deliveryTag)
                        .transferTransactionalState(txnState.getCurrentTransactionId())
                        .transferSettled(Boolean.TRUE)
-                       .transferDeliveryId(UnsignedInteger.valueOf(1))
                        .transfer();
 
-            final Discharge discharge = new Discharge();
-            discharge.setTxnId(txnState.getCurrentTransactionId());
-            discharge.setFail(false);
-            interaction.transferHandle(txnState.getHandle())
-                       .transferDeliveryId(UnsignedInteger.valueOf(2))
-                       .transferSettled(Boolean.FALSE)
-                       .transferDeliveryTag(new Binary(("transaction-" + 2).getBytes(StandardCharsets.UTF_8)))
-                       .transferPayloadData(discharge).transfer();
+            interaction.txnDischarge(txnState, false);
 
-            Disposition dischargeTransactionDisposition =
-                    getDispositionForDeliveryId(interaction, UnsignedInteger.valueOf(2));
+            DeliveryState txDischargeDeliveryState = txnState.getDeliveryState();
+            assertThat(txDischargeDeliveryState, is(instanceOf(Rejected.class)));
 
-            assertThat(dischargeTransactionDisposition.getSettled(), is(equalTo(true)));
-            assertThat(dischargeTransactionDisposition.getState(), is(instanceOf(Rejected.class)));
-
-            Rejected rejected = (Rejected) dischargeTransactionDisposition.getState();
+            Rejected rejected = (Rejected) txDischargeDeliveryState;
             Error error = rejected.getError();
 
             assertThat(error, is(notNullValue()));
@@ -536,14 +526,7 @@
             interaction.begin()
                        .consumeResponse(Begin.class)
 
-                       .attachRole(Role.SENDER)
-                       .attachName("testTransactionCoordinator-" + txnState.getHandle())
-                       .attachHandle(txnState.getHandle())
-                       .attachInitialDeliveryCount(UnsignedInteger.ZERO)
-                       .attachTarget(new Coordinator())
-                       .attachSourceOutcomes(Accepted.ACCEPTED_SYMBOL)
-                       .attach().consumeResponse(Attach.class)
-                       .consumeResponse(Flow.class)
+                       .txnAttachCoordinatorLink(txnState, Accepted.ACCEPTED_SYMBOL)
                        .txnDeclare(txnState)
 
                        .attachRole(Role.SENDER)
@@ -554,22 +537,14 @@
                        .attach().consumeResponse(Attach.class)
                        .consumeResponse(Flow.class)
 
+                       .transferDeliveryId()
                        .transferHandle(linkHandle)
                        .transferPayload(generateMessagePayloadToDestination("Unknown"))
                        .transferDeliveryTag(_deliveryTag)
                        .transferTransactionalState(txnState.getCurrentTransactionId())
                        .transferSettled(Boolean.TRUE)
-                       .transfer();
-
-            final Discharge discharge = new Discharge();
-            discharge.setTxnId(txnState.getCurrentTransactionId());
-            discharge.setFail(false);
-
-            interaction.transferHandle(txnState.getHandle())
-                       .transferSettled(Boolean.FALSE)
-                       .transferDeliveryId(UnsignedInteger.valueOf(4))
-                       .transferDeliveryTag(new Binary(("transaction-" + 4).getBytes(StandardCharsets.UTF_8)))
-                       .transferPayloadData(discharge).transfer();
+                       .transfer()
+                       .txnSendDischarge(txnState, false);
 
             Detach transactionCoordinatorDetach = interaction.consumeResponse().getLatestResponse(Detach.class);
             Error transactionCoordinatorDetachError = transactionCoordinatorDetach.getError();
diff --git a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/extensions/qpid/queue/QueueDeletionTest.java b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/extensions/qpid/queue/QueueDeletionTest.java
index d6eb3d1..39a8a9b 100644
--- a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/extensions/qpid/queue/QueueDeletionTest.java
+++ b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/extensions/qpid/queue/QueueDeletionTest.java
@@ -171,7 +171,7 @@
             assertThat(receivedDetach.getError().getCondition(), is(AmqpError.RESOURCE_DELETED));
             assertThat(receivedDetach.getHandle(), is(equalTo(attach.getHandle())));
 
-            interaction.discharge(txnState, false);
+            interaction.txnSendDischarge(txnState, false);
 
             assertTransactionRollbackOnly(interaction, txnState);
         }
@@ -244,7 +244,7 @@
             assertThat(receivedDetach.getError().getCondition(), is(AmqpError.RESOURCE_DELETED));
             assertThat(receivedDetach.getHandle(), is(equalTo(attach.getHandle())));
 
-            interaction.discharge(txnState, false);
+            interaction.txnSendDischarge(txnState, false);
 
             assertTransactionRollbackOnly(interaction, txnState);
         }
diff --git a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/MessageFormat.java b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/MessageFormat.java
index 1cf1ff0..6ac8e7f 100644
--- a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/MessageFormat.java
+++ b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/MessageFormat.java
@@ -70,7 +70,7 @@
     {
         try (FrameTransport transport = new FrameTransport(_brokerAddress).connect())
         {
-            QpidByteBuffer[] payloads = Utils.splitPayload("testData", 2);
+            QpidByteBuffer[] payloads = Utils.splitPayload(getTestName(), 2);
 
             final Response<?> latestResponse = transport.newInteraction()
                                                         .negotiateProtocol().consumeResponse()
diff --git a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/MultiTransferTest.java b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/MultiTransferTest.java
index ed9e58d..3b421c7 100644
--- a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/MultiTransferTest.java
+++ b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/MultiTransferTest.java
@@ -49,6 +49,7 @@
 import org.apache.qpid.server.protocol.v1_0.type.transport.Flow;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Open;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Role;
+import org.apache.qpid.tests.protocol.ChannelClosedResponse;
 import org.apache.qpid.tests.protocol.v1_0.FrameTransport;
 import org.apache.qpid.tests.protocol.v1_0.Interaction;
 import org.apache.qpid.tests.protocol.SpecificationTest;
@@ -229,9 +230,11 @@
             {
                 payload.dispose();
             }
-            Response<?> latestResponse = interaction.consumeResponse(new Class<?>[] {null}).getLatestResponse();
-            assertThat(latestResponse, is(nullValue()));
+            interaction.consumeResponse(null, Flow.class);
         }
+        String secondMessage = getTestName() + "_2";
+        Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, secondMessage);
+        assertThat(Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(secondMessage)));
     }
 
     @Test
@@ -387,7 +390,7 @@
                 payload.dispose();
             }
 
-            interaction.consumeResponse(Detach.class, End.class, Close.class);
+            interaction.consumeResponse(Detach.class, End.class, Close.class, ChannelClosedResponse.class);
         }
     }
 }
diff --git a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/TransferTest.java b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/TransferTest.java
index 7950741..42f5dfc 100644
--- a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/TransferTest.java
+++ b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/TransferTest.java
@@ -38,6 +38,7 @@
 
 import java.net.InetSocketAddress;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.TreeSet;
@@ -62,7 +63,6 @@
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Header;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Received;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Rejected;
-import org.apache.qpid.server.protocol.v1_0.type.transaction.Discharge;
 import org.apache.qpid.server.protocol.v1_0.type.transport.AmqpError;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Attach;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Begin;
@@ -78,6 +78,7 @@
 import org.apache.qpid.server.protocol.v1_0.type.transport.Role;
 import org.apache.qpid.server.protocol.v1_0.type.transport.SenderSettleMode;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Transfer;
+import org.apache.qpid.tests.protocol.ChannelClosedResponse;
 import org.apache.qpid.tests.protocol.Response;
 import org.apache.qpid.tests.protocol.SpecificationTest;
 import org.apache.qpid.tests.protocol.v1_0.FrameTransport;
@@ -121,13 +122,13 @@
 
     @Test
     @SpecificationTest(section = "1.3.4",
-            description = "Transfer without mandatory fields should result in a decoding error.")
+            description = "mandatory [...] a non null value for the field is always encoded.")
     public void emptyTransfer() throws Exception
     {
         try (FrameTransport transport = new FrameTransport(_brokerAddress).connect())
         {
-            Close responseClose = transport.newInteraction()
-                                           .negotiateProtocol().consumeResponse()
+            Interaction interact = transport.newInteraction();
+            Response<?> response = interact.negotiateProtocol().consumeResponse()
                                            .open().consumeResponse(Open.class)
                                            .begin().consumeResponse(Begin.class)
                                            .attachRole(Role.SENDER)
@@ -136,9 +137,27 @@
                                            .transferHandle(null)
                                            .transfer()
                                            .consumeResponse()
-                                           .getLatestResponse(Close.class);
-            assertThat(responseClose.getError(), is(notNullValue()));
-            assertThat(responseClose.getError().getCondition(), equalTo(AmqpError.DECODE_ERROR));
+                                           .getLatestResponse();
+
+            assertThat(response.getBody(), is(notNullValue()));
+
+            if (response.getBody() instanceof Close)
+            {
+                final Close responseClose = (Close)response.getBody();
+                assertThat(responseClose.getError(), is(notNullValue()));
+                assertThat(responseClose.getError().getCondition(), equalTo(AmqpError.DECODE_ERROR));
+
+                interact.close().sync();
+            }
+            else if (response.getBody() instanceof End)
+            {
+                final End responseEnd = (End)response.getBody();
+                assertThat(responseEnd.getError(), is(notNullValue()));
+                assertThat(responseEnd.getError().getCondition(), equalTo(AmqpError.DECODE_ERROR));
+
+                interact.end().doCloseConnection();
+            }
+            transport.assertNoMoreResponses();
         }
     }
 
@@ -158,10 +177,11 @@
                                                  .attachTargetAddress(BrokerAdmin.TEST_QUEUE_NAME)
                                                  .attach().consumeResponse(Attach.class)
                                                  .consumeResponse(Flow.class)
+                                                 .transferDeliveryId()
                                                  .transferDeliveryTag(null)
                                                  .transferPayloadData(getTestName())
                                                  .transfer();
-            interaction.consumeResponse(Detach.class, End.class, Close.class);
+            interaction.consumeResponse(Detach.class, End.class, Close.class, ChannelClosedResponse.class);
         }
     }
 
@@ -182,6 +202,7 @@
                                                        .attachHandle(linkHandle)
                                                        .attach().consumeResponse(Attach.class)
                                                        .consumeResponse(Flow.class)
+                                                       .transferDeliveryId()
                                                        .transferHandle(linkHandle)
                                                        .transferPayloadData(getTestName())
                                                        .transfer()
@@ -283,12 +304,18 @@
 
     @Test
     @SpecificationTest(section = "2.7.5",
-            description = "If the negotiated link value is first, then it is illegal to set this field to second.")
+            description = "rcv-settle-mode "
+                          + "If first, this indicates that the receiver MUST settle the delivery once it has arrived"
+                          + " without waiting for the sender to settle first."
+                          + " If second, this indicates that the receiver MUST NOT settle until sending its disposition"
+                          + " to the sender and receiving a settled disposition from the sender."
+                          + " If not set, this value is defaulted to the value negotiated on link attach."
+                          + " If the negotiated link value is first, then it is illegal to set this field to second.")
     public void transferReceiverSettleModeCannotBeSecondWhenLinkModeIsFirst() throws Exception
     {
         try (FrameTransport transport = new FrameTransport(_brokerAddress).connect())
         {
-            Detach detach = transport.newInteraction()
+            Response<?> response = transport.newInteraction()
                                      .negotiateProtocol().consumeResponse()
                                      .open().consumeResponse(Open.class)
                                      .begin().consumeResponse(Begin.class)
@@ -301,10 +328,24 @@
                                      .transferRcvSettleMode(ReceiverSettleMode.SECOND)
                                      .transfer()
                                      .consumeResponse()
-                                     .getLatestResponse(Detach.class);
-            Error error = detach.getError();
-            assertThat(error, is(notNullValue()));
-            assertThat(error.getCondition(), is(equalTo(AmqpError.INVALID_FIELD)));
+                                     .getLatestResponse();
+
+            if (response.getBody() instanceof Detach)
+            {
+                final Detach detach = (Detach) response.getBody();
+                Error error = detach.getError();
+                assertThat(error, is(notNullValue()));
+                assertThat(error.getCondition(), is(equalTo(AmqpError.INVALID_FIELD)));
+            }
+            else
+            {
+                if (response.getBody() instanceof Disposition)
+                {
+                    // clean up
+                    Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME);
+                }
+                fail("it is illegal to set transfer 'rcv-settle-mode' to 'second' when link 'rcv-settle-mode' is set to 'first'");
+            }
         }
     }
 
@@ -366,6 +407,7 @@
                                                                                    Rejected.REJECTED_SYMBOL)
                                                              .attach().consumeResponse(Attach.class)
                                                              .consumeResponse(Flow.class)
+                                                             .transferDeliveryId()
                                                              .transferPayload(messageEncoder.getPayload())
                                                              .transferRcvSettleMode(ReceiverSettleMode.FIRST)
                                                              .transfer()
@@ -411,6 +453,7 @@
                                                   .attachSourceOutcomes(Accepted.ACCEPTED_SYMBOL)
                                                   .attach().consumeResponse(Attach.class)
                                                   .consumeResponse(Flow.class)
+                                                  .transferDeliveryId()
                                                   .transferPayload(messageEncoder.getPayload())
                                                   .transferRcvSettleMode(ReceiverSettleMode.FIRST)
                                                   .transfer()
@@ -453,7 +496,7 @@
                                                      .attachSourceAddress(BrokerAdmin.TEST_QUEUE_NAME)
                                                      .attach().consumeResponse()
                                                      .flowIncomingWindow(UnsignedInteger.ONE)
-                                                     .flowNextIncomingId(UnsignedInteger.ZERO)
+                                                     .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
                                                      .flowOutgoingWindow(UnsignedInteger.ZERO)
                                                      .flowNextOutgoingId(UnsignedInteger.ZERO)
                                                      .flowLinkCredit(UnsignedInteger.ONE)
@@ -493,7 +536,7 @@
                                                      .attachRcvSettleMode(ReceiverSettleMode.FIRST)
                                                      .attach().consumeResponse()
                                                      .flowIncomingWindow(UnsignedInteger.ONE)
-                                                     .flowNextIncomingId(UnsignedInteger.ZERO)
+                                                     .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
                                                      .flowOutgoingWindow(UnsignedInteger.ZERO)
                                                      .flowNextOutgoingId(UnsignedInteger.ZERO)
                                                      .flowLinkCredit(UnsignedInteger.ONE)
@@ -519,8 +562,6 @@
     @SpecificationTest(section = "2.6.12", description = "Transferring A Message.")
     public void receiveTransferReceiverSettleSecond() throws Exception
     {
-        Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, getTestName());
-
         try (FrameTransport transport = new FrameTransport(_brokerAddress).connect())
         {
             final Interaction interaction = transport.newInteraction()
@@ -532,19 +573,22 @@
                                                      .attachRcvSettleMode(ReceiverSettleMode.SECOND)
                                                      .attach().consumeResponse()
                                                      .flowIncomingWindow(UnsignedInteger.ONE)
-                                                     .flowNextIncomingId(UnsignedInteger.ZERO)
+                                                     .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
                                                      .flowOutgoingWindow(UnsignedInteger.ZERO)
                                                      .flowNextOutgoingId(UnsignedInteger.ZERO)
                                                      .flowLinkCredit(UnsignedInteger.ONE)
                                                      .flowHandleFromLinkHandle()
-                                                     .flow()
-                                                     .receiveDelivery()
-                                                     .decodeLatestDelivery();
+                                                     .flow();
 
-            Object data = interaction.getDecodedLatestDelivery();
+            Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, getTestName());
+
+            Object data = interaction.receiveDelivery()
+                                     .decodeLatestDelivery()
+                                     .getDecodedLatestDelivery();
             assertThat(data, is(equalTo(getTestName())));
 
             Disposition disposition = interaction.dispositionSettled(false)
+                                                 .dispositionFirstFromLatestDelivery()
                                                  .dispositionRole(Role.RECEIVER)
                                                  .dispositionState(new Accepted())
                                                  .disposition()
@@ -552,8 +596,11 @@
                                                  .getLatestResponse(Disposition.class);
             assertThat(disposition.getSettled(), is(true));
 
-            interaction.consumeResponse(null, Flow.class);
-
+            interaction.dispositionSettled(true)
+                       .dispositionFirstFromLatestDelivery()
+                       .dispositionRole(Role.RECEIVER)
+                       .dispositionState(new Accepted())
+                       .disposition();
         }
     }
 
@@ -561,8 +608,6 @@
     @SpecificationTest(section = "2.6.12", description = "Transferring A Message.")
     public void receiveTransferReceiverSettleSecondWithRejectedOutcome() throws Exception
     {
-        Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, getTestName());
-
         try (FrameTransport transport = new FrameTransport(_brokerAddress).connect())
         {
             final Interaction interaction = transport.newInteraction()
@@ -575,17 +620,20 @@
                                                      .attachRcvSettleMode(ReceiverSettleMode.SECOND)
                                                      .attach().consumeResponse()
                                                      .flowIncomingWindow(UnsignedInteger.ONE)
-                                                     .flowNextIncomingId(UnsignedInteger.ZERO)
+                                                     .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
                                                      .flowOutgoingWindow(UnsignedInteger.ZERO)
                                                      .flowNextOutgoingId(UnsignedInteger.ZERO)
                                                      .flowLinkCredit(UnsignedInteger.ONE)
                                                      .flowHandleFromLinkHandle()
                                                      .flow();
 
+            Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, getTestName());
+
             Object data = interaction.receiveDelivery().decodeLatestDelivery().getDecodedLatestDelivery();
             assertThat(data, is(equalTo(getTestName())));
 
             interaction.dispositionSettled(false)
+                       .dispositionFirstFromLatestDelivery()
                        .dispositionRole(Role.RECEIVER)
                        .dispositionState(new Rejected())
                        .disposition()
@@ -599,9 +647,11 @@
             Disposition disposition = interaction.getLatestResponse(Disposition.class);
             assertThat(disposition.getSettled(), is(true));
 
-            interaction.consumeResponse(null, Flow.class);
-
-
+            interaction.dispositionSettled(true)
+                       .dispositionFirstFromLatestDelivery()
+                       .dispositionRole(Role.RECEIVER)
+                       .dispositionState(new Rejected())
+                       .disposition();
 
         }
         assertThat(Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(getTestName())));
@@ -627,7 +677,7 @@
                                                      .attachSourceDefaultOutcome(null)
                                                      .attach().consumeResponse()
                                                      .flowIncomingWindow(UnsignedInteger.ONE)
-                                                     .flowNextIncomingId(UnsignedInteger.ZERO)
+                                                     .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
                                                      .flowOutgoingWindow(UnsignedInteger.ZERO)
                                                      .flowNextOutgoingId(UnsignedInteger.ZERO)
                                                      .flowLinkCredit(UnsignedInteger.ONE)
@@ -640,6 +690,7 @@
             assertThat(data, is(equalTo(getTestName())));
 
             Disposition disposition = interaction.dispositionSettled(false)
+                                                 .dispositionFirstFromLatestDelivery()
                                                  .dispositionRole(Role.RECEIVER)
                                                  .dispositionState(null)
                                                  .disposition()
@@ -658,7 +709,6 @@
                                                          + " non-terminal delivery states to the sender")
     public void receiveTransferReceiverIndicatesNonTerminalDeliveryState() throws Exception
     {
-        String testMessageData;
         try (FrameTransport transport = new FrameTransport(_brokerAddress).connect())
         {
             final Interaction interaction = transport.newInteraction();
@@ -667,19 +717,13 @@
                                    .openMaxFrameSize(UnsignedInteger.valueOf(4096))
                                    .open().consumeResponse()
                                    .getLatestResponse(Open.class);
-
-            int negotiatedFrameSize = open.getMaxFrameSize().intValue();
-            testMessageData = Stream.generate(() -> "*").limit(negotiatedFrameSize).collect(Collectors.joining());
-
-            Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, testMessageData);
-
             interaction.begin().consumeResponse()
                        .attachRole(Role.RECEIVER)
                        .attachSourceAddress(BrokerAdmin.TEST_QUEUE_NAME)
                        .attachRcvSettleMode(ReceiverSettleMode.SECOND)
                        .attach().consumeResponse()
                        .flowIncomingWindow(UnsignedInteger.ONE)
-                       .flowNextIncomingId(UnsignedInteger.ZERO)
+                       .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
                        .flowOutgoingWindow(UnsignedInteger.ZERO)
                        .flowNextOutgoingId(UnsignedInteger.ZERO)
                        .flowLinkCredit(UnsignedInteger.ONE)
@@ -687,6 +731,11 @@
                        .flow()
                        .sync();
 
+            final int negotiatedFrameSize = open.getMaxFrameSize().intValue();
+            final String testMessageData = Stream.generate(() -> "*").limit(negotiatedFrameSize).collect(Collectors.joining());
+
+            Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, testMessageData);
+
             MessageDecoder messageDecoder = new MessageDecoder();
 
             Transfer first = interaction.consumeResponse(Transfer.class)
@@ -745,14 +794,14 @@
             assumeThat(attach.getSndSettleMode(), is(equalTo(SenderSettleMode.SETTLED)));
 
             interaction.flowIncomingWindow(UnsignedInteger.ONE)
-                                                     .flowNextIncomingId(UnsignedInteger.ZERO)
-                                                     .flowOutgoingWindow(UnsignedInteger.ZERO)
-                                                     .flowNextOutgoingId(UnsignedInteger.ZERO)
-                                                     .flowLinkCredit(UnsignedInteger.ONE)
-                                                     .flowHandleFromLinkHandle()
-                                                     .flow();
+                       .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
+                       .flowOutgoingWindow(UnsignedInteger.ZERO)
+                       .flowNextOutgoingId(UnsignedInteger.ZERO)
+                       .flowLinkCredit(UnsignedInteger.ONE)
+                       .flowHandleFromLinkHandle()
+                       .flow();
 
-            List<Transfer> transfers = interaction.receiveDelivery().getLatestDelivery();
+            final List<Transfer> transfers = interaction.receiveDelivery().getLatestDelivery();
             final AtomicBoolean isSettled = new AtomicBoolean();
             transfers.forEach(transfer -> { if (Boolean.TRUE.equals(transfer.getSettled())) { isSettled.set(true);}});
 
@@ -762,10 +811,6 @@
             interaction.doCloseConnection();
         }
 
-        if (getBrokerAdmin().isQueueDepthSupported())
-        {
-            assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(0)));
-        }
         Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, "test");
         assertThat(Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME), is(equalTo("test")));
     }
@@ -793,17 +838,13 @@
                        .attachTargetAddress(BrokerAdmin.TEST_QUEUE_NAME)
                        .attach()
                        .consumeResponse(Attach.class)
-                       .consumeResponse(Flow.class);
-
-            Flow flow = interaction.getLatestResponse(Flow.class);
-            assertThat(flow.getLinkCredit().intValue(), is(greaterThan(1)));
-
-            interaction.transferDeliveryId(UnsignedInteger.ZERO)
+                       .consumeResponse(Flow.class)
+                       .transferDeliveryId()
                        .transferDeliveryTag(deliveryTag)
                        .transferPayloadData(content1)
                        .transfer()
                        .transferDeliveryTag(deliveryTag)
-                       .transferDeliveryId(UnsignedInteger.ONE)
+                       .transferDeliveryId()
                        .transferPayloadData(getTestName() + "_2")
                        .transfer()
                        .sync();
@@ -1019,28 +1060,21 @@
             interaction.txnAttachCoordinatorLink(txnState)
                        .txnDeclare(txnState);
 
-            interaction.transferDeliveryId(UnsignedInteger.ONE)
+            interaction.transferDeliveryId()
                        .transferDeliveryTag(new Binary("A".getBytes(StandardCharsets.UTF_8)))
                        .transferPayloadData(contents[0])
                        .transfer()
-                       .transferDeliveryId(UnsignedInteger.valueOf(2))
+                       .transferDeliveryId()
                        .transferDeliveryTag(new Binary("B".getBytes(StandardCharsets.UTF_8)))
                        .transferPayloadData(contents[1])
                        .transfer()
-                       .transferDeliveryId(UnsignedInteger.valueOf(3))
+                       .transferDeliveryId()
                        .transferDeliveryTag(new Binary("C".getBytes(StandardCharsets.UTF_8)))
                        .transferTransactionalState(txnState.getCurrentTransactionId())
                        .transferPayloadData(contents[2])
                        .transfer();
 
-            final Discharge discharge = new Discharge();
-            discharge.setTxnId(txnState.getCurrentTransactionId());
-            discharge.setFail(false);
-
-            interaction.transferHandle(txnState.getHandle())
-                       .transferDeliveryId(UnsignedInteger.valueOf(4))
-                       .transferDeliveryTag(new Binary(("transaction-" + 4).getBytes(StandardCharsets.UTF_8)))
-                       .transferPayloadData(discharge).transfer();
+            interaction.txnSendDischarge(txnState, false);
 
             assertDeliveries(interaction, Sets.newTreeSet(Arrays.asList(UnsignedInteger.ONE,
                                                                         UnsignedInteger.valueOf(2),
@@ -1069,39 +1103,32 @@
                                                      .attachRcvSettleMode(ReceiverSettleMode.FIRST)
                                                      .attach().consumeResponse()
                                                      .flowIncomingWindow(UnsignedInteger.valueOf(numberOfMessages))
-                                                     .flowNextIncomingId(UnsignedInteger.ZERO)
+                                                     .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
                                                      .flowOutgoingWindow(UnsignedInteger.ZERO)
                                                      .flowNextOutgoingId(UnsignedInteger.ZERO)
                                                      .flowLinkCredit(UnsignedInteger.valueOf(numberOfMessages))
                                                      .flowHandleFromLinkHandle()
                                                      .flow();
 
-            for (int i = 0; i < contents.length; i++)
+            UnsignedInteger firstDeliveryId = null;
+            for (final String content : contents)
             {
                 interaction.receiveDelivery(Flow.class).decodeLatestDelivery();
                 Object data = interaction.getDecodedLatestDelivery();
-                assertThat(data, is(equalTo(contents[i])));
-                assertThat(interaction.getLatestDeliveryId(), is(equalTo(UnsignedInteger.valueOf(i))));
+                assertThat(data, is(equalTo(content)));
+                if (firstDeliveryId == null)
+                {
+                    firstDeliveryId = interaction.getLatestDeliveryId();
+                }
             }
 
             interaction.dispositionSettled(true)
                        .dispositionRole(Role.RECEIVER)
-                       .dispositionFirst(UnsignedInteger.ZERO)
+                       .dispositionFirst(firstDeliveryId)
                        .dispositionLast(interaction.getLatestDeliveryId())
                        .dispositionState(new Accepted())
                        .disposition();
-
-            // make sure sure the disposition is handled by making drain request
-            interaction.flowLinkCredit(UnsignedInteger.ONE)
-                       .flowNextIncomingId(UnsignedInteger.valueOf(numberOfMessages))
-                       .flowDrain(Boolean.TRUE)
-                       .flow()
-                       .consumeResponse(Flow.class);
-
-            if (getBrokerAdmin().isQueueDepthSupported())
-            {
-                assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(0)));
-            }
+            interaction.doCloseConnection();
         }
 
         final String messageText = getTestName() + "_" + 4;
@@ -1116,7 +1143,6 @@
     {
         final int numberOfMessages = 4;
         final String[] contents = Utils.createTestMessageContents(numberOfMessages, getTestName());
-        Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, contents);
 
         try (FrameTransport transport = new FrameTransport(_brokerAddress).connect())
         {
@@ -1130,19 +1156,22 @@
                                                      .attachHandle(UnsignedInteger.ZERO)
                                                      .attach().consumeResponse()
                                                      .flowIncomingWindow(UnsignedInteger.valueOf(numberOfMessages))
-                                                     .flowNextIncomingId(UnsignedInteger.ZERO)
+                                                     .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
                                                      .flowOutgoingWindow(UnsignedInteger.ZERO)
                                                      .flowNextOutgoingId(UnsignedInteger.ZERO)
                                                      .flowLinkCredit(UnsignedInteger.valueOf(numberOfMessages))
                                                      .flowHandleFromLinkHandle()
                                                      .flow();
 
-            for (int i = 0; i < contents.length; i++)
+            Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, contents);
+
+            final List<UnsignedInteger> deliveryIds = new ArrayList<>();
+            for (final String content : contents)
             {
                 interaction.receiveDelivery(Flow.class).decodeLatestDelivery();
                 Object data = interaction.getDecodedLatestDelivery();
-                assertThat(data, is(equalTo(contents[i])));
-                assertThat(interaction.getLatestDeliveryId(), is(equalTo(UnsignedInteger.valueOf(i))));
+                assertThat(data, is(equalTo(content)));
+                deliveryIds.add(interaction.getLatestDeliveryId());
             }
 
             final InteractionTransactionalState txnState = interaction.createTransactionalState(UnsignedInteger.ONE);
@@ -1151,31 +1180,18 @@
 
             interaction.dispositionSettled(true)
                        .dispositionRole(Role.RECEIVER)
-                       .dispositionFirst(UnsignedInteger.ZERO)
-                       .dispositionLast(UnsignedInteger.ONE)
+                       .dispositionFirst(deliveryIds.get(0))
+                       .dispositionLast(deliveryIds.get(1))
                        .dispositionState(new Accepted())
                        .disposition()
                        .dispositionSettled(true)
                        .dispositionRole(Role.RECEIVER)
-                       .dispositionFirst(UnsignedInteger.valueOf(2))
-                       .dispositionLast(UnsignedInteger.valueOf(3))
+                       .dispositionFirst(deliveryIds.get(2))
+                       .dispositionLast(deliveryIds.get(3))
                        .dispositionTransactionalState(txnState.getCurrentTransactionId(), new Accepted())
                        .disposition();
 
-
-            final Discharge discharge = new Discharge();
-            discharge.setTxnId(txnState.getCurrentTransactionId());
-            discharge.setFail(false);
-
-            interaction.transferHandle(txnState.getHandle())
-                       .transferDeliveryId(UnsignedInteger.valueOf(4))
-                       .transferDeliveryTag(new Binary(("transaction-" + 4).getBytes(StandardCharsets.UTF_8)))
-                       .transferPayloadData(discharge)
-                       .transfer();
-
-
-            final Flow coordinatorFlow = interaction.consume(Flow.class, Disposition.class);
-            assertThat(coordinatorFlow.getHandle(), is(equalTo(txnState.getHandle())));
+            interaction.txnDischarge(txnState, false);
         }
 
         String messageText = getTestName() + "_" + 4;
@@ -1203,6 +1219,10 @@
                               expectedDeliveryIds.remove(deliveryId);
                           });
             }
+            else if (response.getBody() instanceof Flow)
+            {
+                // ignore flows
+            }
         }
         while (!expectedDeliveryIds.isEmpty());
     }
diff --git a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/transaction/DischargeTest.java b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/transaction/DischargeTest.java
index 32a96b0..85040c8 100644
--- a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/transaction/DischargeTest.java
+++ b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/transaction/DischargeTest.java
@@ -21,6 +21,7 @@
 package org.apache.qpid.tests.protocol.v1_0.transaction;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.qpid.tests.utils.BrokerAdmin.KIND_BROKER_J;
 import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.CoreMatchers.not;
 import static org.hamcrest.CoreMatchers.notNullValue;
@@ -28,7 +29,6 @@
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
-import static org.junit.Assume.assumeThat;
 
 import java.net.InetSocketAddress;
 import java.util.List;
@@ -40,10 +40,7 @@
 import org.apache.qpid.server.protocol.v1_0.type.UnsignedInteger;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Accepted;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Rejected;
-import org.apache.qpid.server.protocol.v1_0.type.transaction.Coordinator;
-import org.apache.qpid.server.protocol.v1_0.type.transaction.Declare;
 import org.apache.qpid.server.protocol.v1_0.type.transaction.Declared;
-import org.apache.qpid.server.protocol.v1_0.type.transaction.Discharge;
 import org.apache.qpid.server.protocol.v1_0.type.transaction.TransactionError;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Attach;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Begin;
@@ -55,10 +52,10 @@
 import org.apache.qpid.server.protocol.v1_0.type.transport.ReceiverSettleMode;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Role;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Transfer;
+import org.apache.qpid.tests.protocol.SpecificationTest;
 import org.apache.qpid.tests.protocol.v1_0.FrameTransport;
 import org.apache.qpid.tests.protocol.v1_0.Interaction;
 import org.apache.qpid.tests.protocol.v1_0.InteractionTransactionalState;
-import org.apache.qpid.tests.protocol.SpecificationTest;
 import org.apache.qpid.tests.protocol.v1_0.Utils;
 import org.apache.qpid.tests.utils.BrokerAdmin;
 import org.apache.qpid.tests.utils.BrokerAdminUsingTestBase;
@@ -76,86 +73,76 @@
 
     @Test
     @SpecificationTest(section = "4.3",
-            description = "If the coordinator is unable to complete the discharge, the coordinator MUST convey the error to the controller "
-                          + "as a transaction-error. If the source for the link to the coordinator supports the rejected outcome, then the "
-                          + "message MUST be rejected with this outcome carrying the transaction-error.")
+            description = "If the coordinator is unable to complete the discharge,"
+                          + " the coordinator MUST convey the error to the controller as a transaction-error."
+                          + " If the source for the link to the coordinator supports the rejected outcome, then the "
+                          + " message MUST be rejected with this outcome carrying the transaction-error.")
     public void dischargeUnknownTransactionIdWhenSourceSupportsRejectedOutcome() throws Exception
     {
         try (FrameTransport transport = new FrameTransport(_brokerAddress).connect())
         {
+            final InteractionTransactionalState txnState = new InteractionTransactionalState(UnsignedInteger.ZERO);
             final Interaction interaction = transport.newInteraction();
-            final Disposition disposition = interaction.negotiateProtocol().consumeResponse()
-                                                       .open().consumeResponse(Open.class)
-                                                       .begin().consumeResponse(Begin.class)
-                                                       .attachRole(Role.SENDER)
-                                                       .attachSourceOutcomes(Rejected.REJECTED_SYMBOL)
-                                                       .attachTarget(new Coordinator())
-                                                       .attach().consumeResponse(Attach.class)
-                                                       .consumeResponse(Flow.class)
-                                                       .transferPayloadData(new Declare())
-                                                       .transfer().consumeResponse()
-                                                       .getLatestResponse(Disposition.class);
+            interaction.negotiateProtocol().consumeResponse()
+                       .open().consumeResponse(Open.class)
+                       .begin().consumeResponse(Begin.class)
 
-            assertThat(disposition.getSettled(), is(equalTo(true)));
-            assertThat(disposition.getState(), is(instanceOf(Declared.class)));
-            assertThat(((Declared) disposition.getState()).getTxnId(), is(notNullValue()));
+                       .txnAttachCoordinatorLink(txnState, Rejected.REJECTED_SYMBOL)
+                       .txnDeclare(txnState);
 
-            interaction.consumeResponse(Flow.class);
+            assertThat(txnState.getDeliveryState(), is(instanceOf(Declared.class)));
+            assertThat(txnState.getCurrentTransactionId(), is(notNullValue()));
 
-            final Discharge discharge = new Discharge();
-            discharge.setTxnId(new Binary("nonExistingTransaction".getBytes(UTF_8)));
-            final Disposition dischargeDisposition = interaction.transferDeliveryId(UnsignedInteger.ONE)
-                                                                .transferDeliveryTag(new Binary("discharge".getBytes(UTF_8)))
-                                                                .transferPayloadData(discharge)
-                                                                .transfer().consumeResponse()
-                                                                .getLatestResponse(Disposition.class);
-            assertThat(dischargeDisposition.getState(), is(instanceOf(Rejected.class)));
-            final Error error = ((Rejected) dischargeDisposition.getState()).getError();
+            txnState.setLastTransactionId(new Binary("nonExistingTransaction".getBytes(UTF_8)));
+            interaction.txnDischarge(txnState, false);
+            interaction.doCloseConnection();
+
+            assertThat(txnState.getDeliveryState(), is(instanceOf(Rejected.class)));
+            final Error error = ((Rejected) txnState.getDeliveryState()).getError();
             assertThat(error, is(notNullValue()));
-            assertThat(error.getCondition(), is(equalTo(TransactionError.UNKNOWN_ID)));
+
+            if (KIND_BROKER_J.equals(getBrokerAdmin().getKind()))
+            {
+                assertThat(error.getCondition(), is(equalTo(TransactionError.UNKNOWN_ID)));
+            }
         }
     }
 
     @Test
     @SpecificationTest(section = "4.3",
-            description = "If the coordinator is unable to complete the discharge, the coordinator MUST convey the error to the controller "
-                          + "as a transaction-error. [...] If the source does not support "
-                          + "the rejected outcome, the transactional resource MUST detach the link to the coordinator, with the detach "
-                          + "performative carrying the transaction-error.")
+            description = "If the coordinator is unable to complete the discharge,"
+                          + " the coordinator MUST convey the error to the controller as a transaction-error."
+                          + " [...] If the source does not support the rejected outcome, the transactional resource"
+                          + " MUST detach the link to the coordinator, with the detach performative carrying"
+                          + " the transaction-error.")
     public void dischargeUnknownTransactionIdWhenSourceDoesNotSupportRejectedOutcome() throws Exception
     {
         try (FrameTransport transport = new FrameTransport(_brokerAddress).connect())
         {
+            final InteractionTransactionalState txnState = new InteractionTransactionalState(UnsignedInteger.ZERO);
             final Interaction interaction = transport.newInteraction();
-            final Disposition disposition = interaction.negotiateProtocol().consumeResponse()
-                                                       .open().consumeResponse(Open.class)
-                                                       .begin().consumeResponse(Begin.class)
-                                                       .attachRole(Role.SENDER)
-                                                       .attachSourceOutcomes(Accepted.ACCEPTED_SYMBOL)
-                                                       .attachTarget(new Coordinator())
-                                                       .attach().consumeResponse(Attach.class)
-                                                       .consumeResponse(Flow.class)
-                                                       .transferPayloadData(new Declare())
-                                                       .transfer().consumeResponse()
-                                                       .getLatestResponse(Disposition.class);
+            interaction.negotiateProtocol().consumeResponse()
+                                           .open().consumeResponse(Open.class)
+                                           .begin().consumeResponse(Begin.class)
+                                           .txnAttachCoordinatorLink(txnState, Accepted.ACCEPTED_SYMBOL)
+                                           .txnDeclare(txnState);
 
+            assertThat(txnState.getDeliveryState(), is(instanceOf(Declared.class)));
+            assertThat(txnState.getCurrentTransactionId(), is(notNullValue()));
 
-            assertThat(disposition.getSettled(), is(equalTo(true)));
-            assertThat(disposition.getState(), is(instanceOf(Declared.class)));
-            assertThat(((Declared) disposition.getState()).getTxnId(), is(notNullValue()));
+            txnState.setLastTransactionId(new Binary("nonExistingTransaction".getBytes(UTF_8)));
+            interaction.txnSendDischarge(txnState, false);
 
-            interaction.consumeResponse(Flow.class);
+            final Detach detachResponse = interaction.consumeResponse(Detach.class).getLatestResponse(Detach.class);
+            interaction.doCloseConnection();
 
-            final Discharge discharge = new Discharge();
-            discharge.setTxnId(new Binary("nonExistingTransaction".getBytes(UTF_8)));
-            final Detach detachResponse = interaction.transferDeliveryId(UnsignedInteger.ONE)
-                                                                .transferDeliveryTag(new Binary("discharge".getBytes(UTF_8)))
-                                                                .transferPayloadData(discharge)
-                                                                .transfer().consumeResponse(Detach.class)
-                                                                .getLatestResponse(Detach.class);
-            Error error = detachResponse.getError();
+            final Error error = detachResponse.getError();
             assertThat(error, is(notNullValue()));
-            assertThat(error.getCondition(), is(equalTo(TransactionError.UNKNOWN_ID)));
+
+            if (KIND_BROKER_J.equals(getBrokerAdmin().getKind()))
+            {
+                assertThat(error.getCondition(), is(equalTo(TransactionError.UNKNOWN_ID)));
+            }
         }
     }
 
@@ -167,33 +154,34 @@
                           + " desired transaction identifier and the outcome to be applied upon a successful discharge.")
     public void dischargeSettledAfterReceiverDetach() throws Exception
     {
-        assumeThat(getBrokerAdmin().isQueueDepthSupported(), is(true));
-
-        Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, "test message");
         try (FrameTransport transport = new FrameTransport(_brokerAddress).connect())
         {
             final Interaction interaction = transport.newInteraction();
             final InteractionTransactionalState txnState = interaction.createTransactionalState(UnsignedInteger.ZERO);
-            List<Transfer> transfers = interaction.negotiateProtocol().consumeResponse()
-                                                  .open().consumeResponse(Open.class)
-                                                  .begin().consumeResponse(Begin.class)
+            interaction.negotiateProtocol().consumeResponse()
+                       .open().consumeResponse(Open.class)
+                       .begin().consumeResponse(Begin.class)
 
-                                                  .txnAttachCoordinatorLink(txnState)
-                                                  .txnDeclare(txnState)
+                       .txnAttachCoordinatorLink(txnState)
+                       .txnDeclare(txnState)
 
-                                                  .attachRole(Role.RECEIVER)
-                                                  .attachHandle(UnsignedInteger.ONE)
-                                                  .attachSourceAddress(BrokerAdmin.TEST_QUEUE_NAME)
-                                                  .attachRcvSettleMode(ReceiverSettleMode.FIRST)
-                                                  .attach().consumeResponse(Attach.class)
+                       .attachRole(Role.RECEIVER)
+                       .attachHandle(UnsignedInteger.ONE)
+                       .attachSourceAddress(BrokerAdmin.TEST_QUEUE_NAME)
+                       .attachRcvSettleMode(ReceiverSettleMode.FIRST)
+                       .attach().consumeResponse(Attach.class)
 
-                                                  .flowIncomingWindow(UnsignedInteger.ONE)
-                                                  .flowLinkCredit(UnsignedInteger.ONE)
-                                                  .flowHandleFromLinkHandle()
-                                                  .flow()
+                       .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
+                       .flowNextOutgoingId(UnsignedInteger.ZERO)
+                       .flowOutgoingWindow(UnsignedInteger.ZERO)
+                       .flowIncomingWindow(UnsignedInteger.ONE)
+                       .flowLinkCredit(UnsignedInteger.ONE)
+                       .flowHandleFromLinkHandle()
+                       .flow();
 
-                                                  .receiveDelivery()
-                                                  .getLatestDelivery();
+            Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, getTestName());
+
+            List<Transfer> transfers = interaction.receiveDelivery().getLatestDelivery();
             assertThat(transfers, is(notNullValue()));
             assertThat(transfers, is(not(empty())));
             final UnsignedInteger deliveryId = transfers.get(0).getDeliveryId();
@@ -204,8 +192,15 @@
                        .disposition()
                        .txnDischarge(txnState, false);
 
-            assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(0)));
+            assertThat(txnState.getDeliveryState(), is(instanceOf(Accepted.class)));
+
+            interaction.doCloseConnection();
         }
+
+        String secondMessage = getTestName() + "_2";
+        Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, secondMessage);
+        Object receivedMessage = Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME);
+        assertThat(receivedMessage, is(equalTo(secondMessage)));
     }
 
     @Test
@@ -219,8 +214,6 @@
                           + " was associated.")
     public void dischargeSettledAfterSenderDetach() throws Exception
     {
-        assumeThat(getBrokerAdmin().isQueueDepthSupported(), is(true));
-
         try (FrameTransport transport = new FrameTransport(_brokerAddress).connect())
         {
             final Interaction interaction = transport.newInteraction();
@@ -238,20 +231,21 @@
                        .attach().consumeResponse(Attach.class)
                        .consumeResponse(Flow.class)
 
+                       .transferDeliveryId()
                        .transferTransactionalState(txnState.getCurrentTransactionId())
-                       .transferPayloadData("test message")
+                       .transferPayloadData(getTestName())
                        .transferHandle(UnsignedInteger.ONE)
                        .transfer().consumeResponse(Disposition.class)
 
                        .detachHandle(UnsignedInteger.ONE)
                        .detach().consumeResponse(Detach.class);
-
-            assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(0)));
-
             interaction.txnDischarge(txnState, false);
 
-            assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(1)));
+            assertThat(txnState.getDeliveryState(), is(instanceOf(Accepted.class)));
         }
+
+        final Object receivedMessage = Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME);
+        assertThat(receivedMessage, is(equalTo(getTestName())));
     }
 
     @Test
@@ -262,8 +256,6 @@
                           + " reflect the outcome that was applied.")
     public void dischargeUnsettledAfterSenderClose() throws Exception
     {
-        assumeThat(getBrokerAdmin().isQueueDepthSupported(), is(true));
-
         try (FrameTransport transport = new FrameTransport(_brokerAddress).connect())
         {
             final Interaction interaction = transport.newInteraction();
@@ -282,21 +274,22 @@
                        .attach().consumeResponse(Attach.class)
                        .consumeResponse(Flow.class)
 
+                       .transferDeliveryId()
                        .transferTransactionalState(txnState.getCurrentTransactionId())
-                       .transferPayloadData("test message")
+                       .transferPayloadData(getTestName())
                        .transferHandle(UnsignedInteger.ONE)
                        .transfer().consumeResponse(Disposition.class)
 
                        .detachHandle(UnsignedInteger.ONE)
                        .detachClose(true)
                        .detach().consumeResponse(Detach.class);
-
-            assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(0)));
-
             interaction.txnDischarge(txnState, false);
 
-            assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(1)));
+            assertThat(txnState.getDeliveryState(), is(instanceOf(Accepted.class)));
         }
+
+        final Object receivedMessage = Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME);
+        assertThat(receivedMessage, is(equalTo(getTestName())));
     }
 
 }
diff --git a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/transaction/TransactionalTransferTest.java b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/transaction/TransactionalTransferTest.java
index 853fdd9..f44790f 100644
--- a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/transaction/TransactionalTransferTest.java
+++ b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/transaction/TransactionalTransferTest.java
@@ -26,6 +26,7 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 import static org.junit.Assert.fail;
+import static org.junit.Assume.assumeThat;
 
 import java.net.InetSocketAddress;
 import java.util.Collections;
@@ -102,6 +103,7 @@
                                                          .attach().consumeResponse(Attach.class)
                                                          .consumeResponse(Flow.class)
 
+                                                         .transferDeliveryId()
                                                          .transferHandle(linkHandle)
                                                          .transferPayloadData(getTestName())
                                                          .transferTransactionalState(txnState.getCurrentTransactionId())
@@ -116,9 +118,10 @@
 
             interaction.txnDischarge(txnState, false);
 
-            Object receivedMessage = Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME);
-            assertThat(receivedMessage, is(equalTo(getTestName())));
+            assertThat(txnState.getDeliveryState(), is(instanceOf(Accepted.class)));
         }
+        Object receivedMessage = Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME);
+        assertThat(receivedMessage, is(equalTo(getTestName())));
     }
 
     @Test
@@ -150,6 +153,7 @@
                                                          .attach().consumeResponse(Attach.class)
                                                          .consumeResponse(Flow.class)
 
+                                                         .transferDeliveryId()
                                                          .transferHandle(linkHandle)
                                                          .transferPayloadData(getTestName())
                                                          .transferTransactionalState(txnState.getCurrentTransactionId())
@@ -164,16 +168,11 @@
 
             interaction.txnDischarge(txnState, true);
 
-            if (getBrokerAdmin().isQueueDepthSupported())
-            {
-                assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(0)));
-            }
-            else
-            {
-                final String content = getTestName() + "_2";
-                Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, content);
-                assertThat(Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(content)));
-            }
+            assertThat(txnState.getDeliveryState(), is(instanceOf(Accepted.class)));
+
+            final String content = getTestName() + "_2";
+            Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, content);
+            assertThat(Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(content)));
         }
     }
 
@@ -204,9 +203,11 @@
                                                          .attachTargetAddress(BrokerAdmin.TEST_QUEUE_NAME)
                                                          .attachRcvSettleMode(ReceiverSettleMode.SECOND)
                                                          .attachHandle(linkHandle)
-                                                         .attach().consumeResponse(Attach.class)
+                                                         .attach()
+                                                         .consumeResponse(Attach.class)
                                                          .consumeResponse(Flow.class)
 
+                                                         .transferDeliveryId()
                                                          .transferHandle(linkHandle)
                                                          .transferPayloadData(getTestName())
                                                          .transferTransactionalState(txnState.getCurrentTransactionId())
@@ -225,6 +226,8 @@
                        .disposition();
 
             interaction.txnDischarge(txnState, false);
+
+            assertThat(txnState.getDeliveryState(), is(instanceOf(Accepted.class)));
         }
         assertThat(Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(getTestName())));
     }
@@ -255,6 +258,7 @@
                                               .attach().consumeResponse(Attach.class)
                                               .consumeResponse(Flow.class)
 
+                                              .transferDeliveryId()
                                               .transferHandle(linkHandle)
                                               .transferPayloadData(getTestName())
                                               .transferTransactionalState(integerToBinary(Integer.MAX_VALUE))
@@ -295,7 +299,7 @@
                        .consumeResponse(Attach.class)
 
                        .flowIncomingWindow(UnsignedInteger.ONE)
-                       .flowNextIncomingId(UnsignedInteger.ZERO)
+                       .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
                        .flowOutgoingWindow(UnsignedInteger.ZERO)
                        .flowNextOutgoingId(UnsignedInteger.ZERO)
                        .flowLinkCredit(UnsignedInteger.ONE)
@@ -313,6 +317,9 @@
                        .dispositionTransactionalState(txnState.getCurrentTransactionId(), new Accepted())
                        .disposition()
                        .txnDischarge(txnState, false);
+
+
+            assertThat(txnState.getDeliveryState(), is(instanceOf(Accepted.class)));
         }
     }
 
@@ -344,7 +351,7 @@
                        .consumeResponse(Attach.class)
 
                        .flowIncomingWindow(UnsignedInteger.ONE)
-                       .flowNextIncomingId(UnsignedInteger.ZERO)
+                       .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
                        .flowOutgoingWindow(UnsignedInteger.ZERO)
                        .flowNextOutgoingId(UnsignedInteger.ZERO)
                        .flowLinkCredit(UnsignedInteger.ONE)
@@ -363,6 +370,8 @@
                        .disposition()
                        .txnDischarge(txnState, true);
 
+            assertThat(txnState.getDeliveryState(), is(instanceOf(Accepted.class)));
+
             Object receivedMessage = Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME);
             assertThat(receivedMessage, is(equalTo(getTestName())));
         }
@@ -396,7 +405,7 @@
                                                   .attach().consumeResponse(Attach.class)
 
                                                   .flowIncomingWindow(UnsignedInteger.ONE)
-                                                  .flowNextIncomingId(UnsignedInteger.ZERO)
+                                                  .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
                                                   .flowOutgoingWindow(UnsignedInteger.ZERO)
                                                   .flowNextOutgoingId(UnsignedInteger.ZERO)
                                                   .flowLinkCredit(UnsignedInteger.ONE)
@@ -421,7 +430,10 @@
                                               .consumeResponse().getLatestResponse();
             assertUnknownTransactionIdError(response);
         }
-        assertThat(Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(getTestName())));
+        finally
+        {
+            assertThat(Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(getTestName())));
+        }
     }
 
     @Ignore("TODO disposition is currently not being sent by Broker")
@@ -479,6 +491,8 @@
             assertThat(((TransactionalState) settledDisposition.getState()).getOutcome(), is(instanceOf(Accepted.class)));
 
             interaction.txnDischarge(txnState, false);
+
+            assertThat(txnState.getDeliveryState(), is(instanceOf(Accepted.class)));
         }
     }
 
@@ -514,7 +528,7 @@
                        .consumeResponse(Attach.class)
 
                        .flowIncomingWindow(UnsignedInteger.ONE)
-                       .flowNextIncomingId(UnsignedInteger.ZERO)
+                       .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
                        .flowOutgoingWindow(UnsignedInteger.ZERO)
                        .flowNextOutgoingId(UnsignedInteger.ZERO)
                        .flowLinkCredit(UnsignedInteger.ONE)
@@ -526,9 +540,6 @@
 
             List<Transfer> transfers = interaction.getLatestDelivery();
             assertThat(transfers.size(), is(equalTo(1)));
-            Transfer transfer = transfers.get(0);
-            assertThat(transfer.getState(), is(instanceOf(TransactionalState.class)));
-            assertThat(((TransactionalState) transfer.getState()).getTxnId(), is(equalTo(txnState.getCurrentTransactionId())));
 
             Object data = interaction.decodeLatestDelivery().getDecodedLatestDelivery();
             assertThat(data, is(equalTo(getTestName())));
@@ -536,19 +547,20 @@
             interaction.dispositionSettled(true)
                        .dispositionRole(Role.RECEIVER)
                        .dispositionTransactionalState(txnState.getCurrentTransactionId(), new Accepted())
+                       .dispositionFirstFromLatestDelivery()
                        .disposition()
                        .txnDischarge(txnState, false);
 
-            if (getBrokerAdmin().isQueueDepthSupported())
-            {
-                assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(0)));
-            }
-            else
-            {
-                final String content = getTestName() + "_2";
-                Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, content);
-                assertThat(Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(content)));
-            }
+
+            assertThat(txnState.getDeliveryState(), is(instanceOf(Accepted.class)));
+
+            Transfer transfer = transfers.get(0);
+            assumeThat(transfer.getState(), is(instanceOf(TransactionalState.class)));
+            assumeThat(((TransactionalState) transfer.getState()).getTxnId(), is(equalTo(txnState.getCurrentTransactionId())));
+
+            final String content = getTestName() + "_2";
+            Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, content);
+            assertThat(Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(content)));
         }
     }
 
@@ -584,7 +596,7 @@
                        .consumeResponse(Attach.class)
 
                        .flowIncomingWindow(UnsignedInteger.ONE)
-                       .flowNextIncomingId(UnsignedInteger.ZERO)
+                       .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
                        .flowOutgoingWindow(UnsignedInteger.ZERO)
                        .flowNextOutgoingId(UnsignedInteger.ZERO)
                        .flowLinkCredit(UnsignedInteger.ONE)
@@ -596,9 +608,6 @@
 
             List<Transfer> transfers = interaction.getLatestDelivery();
             assertThat(transfers.size(), is(equalTo(1)));
-            Transfer transfer = transfers.get(0);
-            assertThat(transfer.getState(), is(instanceOf(TransactionalState.class)));
-            assertThat(((TransactionalState) transfer.getState()).getTxnId(), is(equalTo(txnState.getCurrentTransactionId())));
 
             Object data = interaction.decodeLatestDelivery().getDecodedLatestDelivery();
             assertThat(data, is(equalTo(getTestName())));
@@ -609,11 +618,13 @@
                        .disposition()
                        .txnDischarge(txnState, true);
 
-            if (getBrokerAdmin().isQueueDepthSupported())
-            {
-                assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(1)));
-            }
+            assertThat(txnState.getDeliveryState(), is(instanceOf(Accepted.class)));
+
             assertThat(Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(getTestName())));
+
+            Transfer transfer = transfers.get(0);
+            assumeThat(transfer.getState(), is(instanceOf(TransactionalState.class)));
+            assumeThat(((TransactionalState) transfer.getState()).getTxnId(), is(equalTo(txnState.getCurrentTransactionId())));
         }
     }
 
@@ -661,7 +672,10 @@
 
             assertUnknownTransactionIdError(response);
         }
-        assertThat(Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(getTestName())));
+        finally
+        {
+            assertThat(Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(getTestName())));
+        }
     }
 
     private void assertUnknownTransactionIdError(final Response<?> response)
diff --git a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/transport/link/FlowTest.java b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/transport/link/FlowTest.java
index 1bfa6d3..448de42 100644
--- a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/transport/link/FlowTest.java
+++ b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/transport/link/FlowTest.java
@@ -21,6 +21,7 @@
 package org.apache.qpid.tests.protocol.v1_0.transport.link;
 
 import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.CoreMatchers.notNullValue;
 import static org.hamcrest.CoreMatchers.nullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -42,6 +43,7 @@
 import org.apache.qpid.server.protocol.v1_0.type.transport.Open;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Role;
 import org.apache.qpid.server.protocol.v1_0.type.transport.SessionError;
+import org.apache.qpid.tests.protocol.Response;
 import org.apache.qpid.tests.protocol.SpecificationTest;
 import org.apache.qpid.tests.protocol.v1_0.FrameTransport;
 import org.apache.qpid.tests.protocol.v1_0.Interaction;
@@ -53,7 +55,7 @@
 {
     @Test
     @SpecificationTest(section = "1.3.4",
-            description = "Flow without mandatory fields should result in a decoding error.")
+            description = "mandatory [...] a non null value for the field is always encoded.")
     public void emptyFlow() throws Exception
     {
         getBrokerAdmin().createQueue(BrokerAdmin.TEST_QUEUE_NAME);
@@ -69,7 +71,7 @@
                                            .flowOutgoingWindow(null)
                                            .flowNextOutgoingId(null)
                                            .flow()
-                                           .consumeResponse()
+                                           .consumeResponse(Close.class)
                                            .getLatestResponse(Close.class);
             assertThat(responseClose.getError(), is(notNullValue()));
             assertThat(responseClose.getError().getCondition(), is(AmqpError.DECODE_ERROR));
@@ -89,6 +91,11 @@
                                          .open().consumeResponse(Open.class)
                                          .begin().consumeResponse(Begin.class)
                                          .flowEcho(true)
+                                         .flowOutgoingWindow(UnsignedInteger.ZERO)
+                                         .flowNextOutgoingId(UnsignedInteger.ZERO)
+                                         .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
+                                         .flowIncomingWindow(UnsignedInteger.ONE)
+                                         .flowHandle(null)
                                          .flow()
                                          .consumeResponse()
                                          .getLatestResponse(Flow.class);
@@ -118,7 +125,10 @@
                                          .flowHandleFromLinkHandle()
                                          .flowAvailable(UnsignedInteger.valueOf(10))
                                          .flowDeliveryCount(UnsignedInteger.ZERO)
-                                         .flowLinkCredit(UnsignedInteger.ZERO)
+                                         .flowLinkCredit(UnsignedInteger.ONE)
+                                         .flowOutgoingWindow(UnsignedInteger.ZERO)
+                                         .flowNextOutgoingId(UnsignedInteger.ZERO)
+                                         .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
                                          .flow().consumeResponse()
                                          .getLatestResponse(Flow.class);
             assertThat(responseFlow.getEcho(), not(equalTo(Boolean.TRUE)));
@@ -146,7 +156,7 @@
                        .attachSourceAddress(BrokerAdmin.TEST_QUEUE_NAME)
                        .attach().consumeResponse()
                        .flowIncomingWindow(UnsignedInteger.ONE)
-                       .flowNextIncomingId(UnsignedInteger.ZERO)
+                       .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
                        .flowOutgoingWindow(UnsignedInteger.ZERO)
                        .flowNextOutgoingId(UnsignedInteger.ZERO)
                        .flowLinkCredit(UnsignedInteger.ONE)
@@ -156,7 +166,7 @@
                        .decodeLatestDelivery()
                        .dispositionSettled(true)
                        .dispositionRole(Role.RECEIVER)
-                       .dispositionFirst(interaction.getLatestDeliveryId())
+                       .dispositionFirstFromLatestDelivery()
                        .dispositionLast(interaction.getLatestDeliveryId())
                        .dispositionState(new Accepted())
                        .disposition()
@@ -215,7 +225,12 @@
                                        .open().consumeResponse(Open.class)
                                        .begin().consumeResponse(Begin.class)
                                        .flowEcho(true)
-                                       .flowHandle(UnsignedInteger.ONE)
+                                       .flowIncomingWindow(UnsignedInteger.ONE)
+                                       .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
+                                       .flowLinkCredit(UnsignedInteger.ONE)
+                                       .flowHandle(UnsignedInteger.valueOf(Integer.MAX_VALUE))
+                                       .flowOutgoingWindow(UnsignedInteger.ZERO)
+                                       .flowNextOutgoingId(UnsignedInteger.ZERO)
                                        .flow()
                                        .consumeResponse().getLatestResponse(End.class);
 
@@ -249,20 +264,22 @@
             UnsignedInteger remoteHandle = remoteAttach.getHandle();
             assertThat(remoteHandle, is(notNullValue()));
 
-            Flow responseFlow = interaction.flowIncomingWindow(UnsignedInteger.valueOf(1))
-                                           .flowNextIncomingId(UnsignedInteger.ZERO)
+            interaction.flowIncomingWindow(UnsignedInteger.ONE)
+                                           .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
+                                           .flowOutgoingWindow(UnsignedInteger.ZERO)
+                                           .flowNextOutgoingId(UnsignedInteger.ZERO)
                                            .flowLinkCredit(UnsignedInteger.ONE)
                                            .flowDrain(Boolean.FALSE)
                                            .flowEcho(Boolean.TRUE)
                                            .flowHandleFromLinkHandle()
                                            .flow()
-                                           .consumeResponse().getLatestResponse(Flow.class);
+                                           .consumeResponse(null, Flow.class);
 
-            assertThat(responseFlow.getHandle(), is(equalTo(remoteHandle)));
-            assertThat(responseFlow.getLinkCredit(), is(equalTo(UnsignedInteger.ONE)));
-            assertThat(responseFlow.getDrain(), is(equalTo(Boolean.FALSE)));
-
-            responseFlow = interaction.flowLinkCredit(UnsignedInteger.ONE)
+            Flow responseFlow = interaction.flowIncomingWindow(UnsignedInteger.ONE)
+                                      .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
+                                      .flowOutgoingWindow(UnsignedInteger.ZERO)
+                                      .flowNextOutgoingId(UnsignedInteger.ZERO)
+                                      .flowLinkCredit(UnsignedInteger.ONE)
                                       .flowDrain(Boolean.TRUE)
                                       .flowEcho(Boolean.FALSE)
                                       .flowHandleFromLinkHandle()
@@ -271,6 +288,7 @@
 
             assertThat(responseFlow.getHandle(), is(equalTo(remoteHandle)));
             assertThat(responseFlow.getLinkCredit(), is(equalTo(UnsignedInteger.ZERO)));
+            assertThat(responseFlow.getDrain(), is(equalTo(Boolean.TRUE)));
         }
     }
 
@@ -286,8 +304,6 @@
     {
         BrokerAdmin brokerAdmin = getBrokerAdmin();
         brokerAdmin.createQueue(BrokerAdmin.TEST_QUEUE_NAME);
-        String messageContent = getTestName();
-        Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, messageContent);
 
         final InetSocketAddress addr = brokerAdmin.getBrokerAddress(BrokerAdmin.PortType.ANONYMOUS_AMQP);
         try (FrameTransport transport = new FrameTransport(addr).connect())
@@ -304,35 +320,42 @@
             UnsignedInteger remoteHandle = remoteAttach.getHandle();
             assertThat(remoteHandle, is(notNullValue()));
 
-            Object receivedMessageContent = interaction.flowIncomingWindow(UnsignedInteger.valueOf(1))
-                                                       .flowNextIncomingId(UnsignedInteger.ZERO)
+            interaction.flowIncomingWindow(UnsignedInteger.valueOf(1))
+                                                       .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
                                                        .flowLinkCredit(UnsignedInteger.ONE)
                                                        .flowDrain(Boolean.FALSE)
                                                        .flowEcho(Boolean.FALSE)
                                                        .flowHandleFromLinkHandle()
+                                                       .flowOutgoingWindow(UnsignedInteger.ZERO)
+                                                       .flowNextOutgoingId(UnsignedInteger.ZERO)
                                                        .flow()
-                                                       .receiveDelivery()
-                                                       .decodeLatestDelivery()
-                                                       .getDecodedLatestDelivery();
+                                                       .sync();
 
-            assertThat(receivedMessageContent, is(equalTo(messageContent)));
-            UnsignedInteger firstDeliveryId = interaction.getLatestDeliveryId();
-            assertThat(firstDeliveryId, is(equalTo(UnsignedInteger.ZERO)));
+            Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, getTestName());
 
-            Flow responseFlow = interaction.flowNextIncomingId(UnsignedInteger.ONE)
-                                           .flowLinkCredit(UnsignedInteger.ONE)
-                                           .flowDrain(Boolean.TRUE)
-                                           .flowEcho(Boolean.FALSE)
-                                           .flowHandleFromLinkHandle()
-                                           .flow()
-                                           .consumeResponse().getLatestResponse(Flow.class);
+            final Object receivedMessageContent = interaction.receiveDelivery()
+                                                             .decodeLatestDelivery()
+                                                             .getDecodedLatestDelivery();
+
+            assertThat(receivedMessageContent, is(equalTo(getTestName())));
+
+            final Flow responseFlow = interaction.flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
+                                                 .flowLinkCredit(UnsignedInteger.ONE)
+                                                 .flowDrain(Boolean.TRUE)
+                                                 .flowEcho(Boolean.FALSE)
+                                                 .flowHandleFromLinkHandle()
+                                                 .flowOutgoingWindow(UnsignedInteger.ZERO)
+                                                 .flowNextOutgoingId(UnsignedInteger.ZERO)
+                                                 .flowDeliveryCount()
+                                                 .flow()
+                                                 .consumeResponse().getLatestResponse(Flow.class);
 
             assertThat(responseFlow.getHandle(), is(equalTo(remoteHandle)));
             assertThat(responseFlow.getLinkCredit(), is(equalTo(UnsignedInteger.ZERO)));
 
             interaction.dispositionSettled(true)
                        .dispositionRole(Role.RECEIVER)
-                       .dispositionFirst(firstDeliveryId)
+                       .dispositionFirst(interaction.getLatestDeliveryId())
                        .dispositionState(new Accepted())
                        .disposition()
                        .sync();
@@ -373,9 +396,12 @@
             UnsignedInteger delta = UnsignedInteger.ONE;
             UnsignedInteger incomingWindow = UnsignedInteger.valueOf(3);
             Object receivedMessageContent1 = interaction.flowIncomingWindow(incomingWindow)
-                                                        .flowNextIncomingId(UnsignedInteger.ZERO)
+                                                        .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
                                                         .flowLinkCredit(delta)
                                                         .flowHandleFromLinkHandle()
+                                                        .flowDeliveryCount()
+                                                        .flowOutgoingWindow(UnsignedInteger.ZERO)
+                                                        .flowNextOutgoingId(UnsignedInteger.ZERO)
                                                         .flow()
                                                         .receiveDelivery()
                                                         .decodeLatestDelivery()
@@ -383,12 +409,14 @@
 
             assertThat(receivedMessageContent1, is(equalTo(contents[0])));
             UnsignedInteger firstDeliveryId = interaction.getLatestDeliveryId();
-            assertThat(firstDeliveryId, is(equalTo(UnsignedInteger.ZERO)));
 
             Object receivedMessageContent2 = interaction.flowIncomingWindow(incomingWindow)
-                                                        .flowNextIncomingId(UnsignedInteger.ONE)
+                                                        .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
                                                         .flowLinkCredit(delta)
                                                         .flowHandleFromLinkHandle()
+                                                        .flowOutgoingWindow(UnsignedInteger.ZERO)
+                                                        .flowNextOutgoingId(UnsignedInteger.ZERO)
+                                                        .flowDeliveryCount()
                                                         .flow()
                                                         .receiveDelivery()
                                                         .decodeLatestDelivery()
@@ -396,17 +424,18 @@
 
             assertThat(receivedMessageContent2, is(equalTo(contents[1])));
             UnsignedInteger secondDeliveryId = interaction.getLatestDeliveryId();
-            assertThat(secondDeliveryId, is(equalTo(UnsignedInteger.ONE)));
 
             // send session flow with echo=true to verify that no message is delivered without issuing a credit
-            Flow responseFlow = interaction.flowNextIncomingId(UnsignedInteger.valueOf(2))
-                                           .flowLinkCredit(null)
-                                           .flowHandle(null)
-                                           .flowEcho(Boolean.TRUE)
-                                           .flow()
-                                           .consumeResponse().getLatestResponse(Flow.class);
-
-            assertThat(responseFlow.getHandle(), is(nullValue()));
+            interaction.flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
+                       .flowIncomingWindow(incomingWindow)
+                       .flowLinkCredit(null)
+                       .flowHandle(null)
+                       .flowDeliveryCount(null)
+                       .flowEcho(Boolean.TRUE)
+                       .flowOutgoingWindow(UnsignedInteger.ZERO)
+                       .flowNextOutgoingId(UnsignedInteger.ZERO)
+                       .flow()
+                       .consumeResponse(null, Flow.class);
 
             interaction.dispositionSettled(true)
                        .dispositionRole(Role.RECEIVER)
@@ -429,10 +458,8 @@
     {
         BrokerAdmin brokerAdmin = getBrokerAdmin();
         brokerAdmin.createQueue(BrokerAdmin.TEST_QUEUE_NAME);
-        final String[] contents = Utils.createTestMessageContents(2, getTestName());
-        Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, contents);
-
         final InetSocketAddress addr = brokerAdmin.getBrokerAddress(BrokerAdmin.PortType.ANONYMOUS_AMQP);
+        Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, getTestName());
         try (FrameTransport transport = new FrameTransport(addr).connect())
         {
             Interaction interaction = transport.newInteraction()
@@ -447,36 +474,68 @@
             UnsignedInteger remoteHandle = remoteAttach.getHandle();
             assertThat(remoteHandle, is(notNullValue()));
 
-            Object receivedMessageContent1 = interaction.flowIncomingWindow(UnsignedInteger.valueOf(2))
-                                                        .flowNextIncomingId(UnsignedInteger.ZERO)
-                                                        .flowLinkCredit(UnsignedInteger.ONE)
+            UnsignedInteger incomingWindow = UnsignedInteger.valueOf(2);
+            Object receivedMessageContent1 = interaction.flowIncomingWindow(incomingWindow)
+                                                        .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
+                                                        .flowLinkCredit(incomingWindow)
+                                                        .flowNextOutgoingId()
                                                         .flowHandleFromLinkHandle()
+                                                        .flowOutgoingWindow(UnsignedInteger.ZERO)
+                                                        .flowNextOutgoingId(UnsignedInteger.ZERO)
                                                         .flow()
                                                         .receiveDelivery()
                                                         .decodeLatestDelivery()
                                                         .getDecodedLatestDelivery();
+            assertThat(receivedMessageContent1, is(equalTo(getTestName())));
 
-            assertThat(receivedMessageContent1, is(equalTo(contents[0])));
-            assertThat(interaction.getLatestDeliveryId(), is(equalTo(UnsignedInteger.ZERO)));
+            final Response<?> response = interaction.flowIncomingWindow(incomingWindow)
+                                                    .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
+                                                    .flowLinkCredit(UnsignedInteger.ZERO)
+                                                    .flowHandleFromLinkHandle()
+                                                    .flowEcho(Boolean.TRUE)
+                                                    .flowOutgoingWindow(UnsignedInteger.ZERO)
+                                                    .flowNextOutgoingId(UnsignedInteger.ZERO)
+                                                    .flowDeliveryCount()
+                                                    .flow()
+                                                    .consumeResponse(null, Flow.class)
+                                                    .getLatestResponse();
 
-            Flow responseFlow = interaction.flowNextIncomingId(UnsignedInteger.ONE)
-                                           .flowLinkCredit(UnsignedInteger.ZERO)
-                                           .flowHandleFromLinkHandle()
-                                           .flowEcho(Boolean.TRUE)
-                                           .flow()
-                                           .consumeResponse().getLatestResponse(Flow.class);
+            if (response != null)
+            {
+                assertThat(response.getBody(), is(instanceOf(Flow.class)));
+                final Flow responseFlow = (Flow) response.getBody();
+                assertThat(responseFlow.getEcho(), not(equalTo(Boolean.TRUE)));
+                assertThat(responseFlow.getHandle(), is(notNullValue()));
+            }
 
-            assertThat(responseFlow.getHandle(), is(equalTo(remoteHandle)));
-            assertThat(responseFlow.getLinkCredit(), is(equalTo(UnsignedInteger.ZERO)));
+            final String message2 = getTestName() + "_2";
+            Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, message2);
+            try
+            {
+                // send session flow with echo=true to verify that no message is delivered without issuing a credit
+                interaction.flowIncomingWindow(incomingWindow)
+                           .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
+                           .flowLinkCredit(null)
+                           .flowHandle(null)
+                           .flowDeliveryCount(null)
+                           .flowEcho(Boolean.TRUE)
+                           .flowOutgoingWindow(UnsignedInteger.ZERO)
+                           .flowNextOutgoingId(UnsignedInteger.ZERO)
+                           .flow()
+                           .consumeResponse(null, Flow.class);
+            }
+            finally
+            {
+                assertThat(Utils.receiveMessage(addr, BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(message2)));
 
-            interaction.dispositionSettled(true)
-                       .dispositionRole(Role.RECEIVER)
-                       .dispositionFirst(interaction.getLatestDeliveryId())
-                       .dispositionState(new Accepted())
-                       .disposition()
-                       .sync();
+                interaction.dispositionSettled(true)
+                           .dispositionRole(Role.RECEIVER)
+                           .dispositionFirst(interaction.getLatestDeliveryId())
+                           .dispositionState(new Accepted())
+                           .disposition()
+                           .sync();
+            }
         }
-        assertThat(Utils.receiveMessage(addr, BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(contents[1])));
     }
 
     @Test
@@ -485,7 +544,7 @@
                           + " available to consume the current link-credit. If set, the sender will"
                           + " (after sending all available messages) advance the delivery-count as much as possible,"
                           + " consuming all link-credit, and send the flow state to the receiver.")
-    public void drainWithZeroCredits() throws Exception
+    public void drain() throws Exception
     {
         BrokerAdmin brokerAdmin = getBrokerAdmin();
         brokerAdmin.createQueue(BrokerAdmin.TEST_QUEUE_NAME);
@@ -507,16 +566,26 @@
             assertThat(remoteHandle, is(notNullValue()));
 
             Flow responseFlow = interaction.flowIncomingWindow(UnsignedInteger.valueOf(2))
-                                           .flowNextIncomingId(UnsignedInteger.ZERO)
-                                           .flowLinkCredit(UnsignedInteger.ZERO)
+                                           .flowNextIncomingIdFromPeerLatestSessionBeginAndDeliveryCount()
+                                           .flowLinkCredit(UnsignedInteger.valueOf(2))
                                            .flowDrain(Boolean.TRUE)
                                            .flowHandleFromLinkHandle()
+                                           .flowOutgoingWindow(UnsignedInteger.ZERO)
+                                           .flowNextOutgoingId(UnsignedInteger.ZERO)
                                            .flow()
-                                           .consumeResponse().getLatestResponse(Flow.class);
+                                           .receiveDelivery()
+                                           .decodeLatestDelivery()
+                                           .consumeResponse(Flow.class).getLatestResponse(Flow.class);
 
             assertThat(responseFlow.getHandle(), is(equalTo(remoteHandle)));
             assertThat(responseFlow.getLinkCredit(), is(equalTo(UnsignedInteger.ZERO)));
+
+            interaction.dispositionSettled(true)
+                       .dispositionRole(Role.RECEIVER)
+                       .dispositionFirstFromLatestDelivery()
+                       .dispositionState(new Accepted())
+                       .disposition()
+                       .sync();
         }
-        assertThat(Utils.receiveMessage(addr, BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(getTestName())));
     }
 }
diff --git a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/transport/link/ResumeDeliveriesTest.java b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/transport/link/ResumeDeliveriesTest.java
index 337c129..61884e3 100644
--- a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/transport/link/ResumeDeliveriesTest.java
+++ b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/transport/link/ResumeDeliveriesTest.java
@@ -591,11 +591,9 @@
 
             interaction.doCloseConnection();
 
-            if (getBrokerAdmin().isQueueDepthSupported())
-            {
-                assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME),
-                           is(equalTo(0)));
-            }
+            final String content = getTestName() + "_2";
+            Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, content);
+            assertThat(Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME), Matchers.is(Matchers.equalTo(content)));
         }
     }
 
@@ -675,11 +673,9 @@
 
             interaction.doCloseConnection();
 
-            if (getBrokerAdmin().isQueueDepthSupported())
-            {
-                assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME),
-                           Matchers.is(Matchers.equalTo(0)));
-            }
+            final String content = getTestName() + "_2";
+            Utils.putMessageOnQueue(getBrokerAdmin(), BrokerAdmin.TEST_QUEUE_NAME, content);
+            assertThat(Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME), Matchers.is(Matchers.equalTo(content)));
         }
     }
 
diff --git a/systests/protocol-tests-core/src/main/java/org/apache/qpid/tests/protocol/AbstractInteraction.java b/systests/protocol-tests-core/src/main/java/org/apache/qpid/tests/protocol/AbstractInteraction.java
index f9640dd..801033e 100644
--- a/systests/protocol-tests-core/src/main/java/org/apache/qpid/tests/protocol/AbstractInteraction.java
+++ b/systests/protocol-tests-core/src/main/java/org/apache/qpid/tests/protocol/AbstractInteraction.java
@@ -44,9 +44,14 @@
 
     public I consumeResponse(final Class<?>... responseTypes) throws Exception
     {
+        final Set<Class<?>> acceptableResponseClasses = new HashSet<>(Arrays.asList(responseTypes));
+        return consumeResponse(acceptableResponseClasses);
+    }
+
+    protected I consumeResponse(final Set<Class<?>> acceptableResponseClasses) throws Exception
+    {
         sync();
         _latestResponse = getNextResponse();
-        final Set<Class<?>> acceptableResponseClasses = new HashSet<>(Arrays.asList(responseTypes));
         if ((acceptableResponseClasses.isEmpty() && _latestResponse != null)
             || (acceptableResponseClasses.contains(null) && _latestResponse == null))
         {