fix one flaky and logs for other

Azure · Mar 7, 2023 · 0e024e0 · 0e024e0
1 parent 3f9a3cb
commit 0e024e0
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 5 deletions.
diff --git a/...g-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventPositionIntegrationTest.java b/...g-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventPositionIntegrationTest.java
@@ -177,7 +177,14 @@ void receiveLatestMessages() {
             StepVerifier.create(consumer.receiveFromPartition(testData.getPartitionId(), EventPosition.latest())
                 .filter(event -> isMatchingEvent(event, messageId))
                 .take(numberOfEvents))
-                .then(() -> producer.send(events, options))
+                .then(() -> {
+                    try {
+                        producer.send(events, options);
+                        logger.atInfo().log("sent events");
+                    } catch (RuntimeException ex) {
+                        throw logger.logThrowableAsError(ex);
+                    }
+                })
                 .expectNextCount(numberOfEvents)
                 .expectComplete()
                 .verify(TIMEOUT);
@@ -194,12 +201,15 @@ void receiveLatestMessages() {
     @Test
     void receiveMessageFromEnqueuedTime() {
         // Arrange
+        final EventData expectedEvent = receivedEvents[0];
+        final Instant enqueuedTime = expectedEvent.getEnqueuedTime();
+
         logger.atInfo()
             .addKeyValue("partitionId", testData.getPartitionId())
-            .addKeyValue("from", testData.getPartitionProperties().getLastEnqueuedTime())
+            .addKeyValue("from", enqueuedTime)
             .log("Receiving events");
-        final EventPosition position = EventPosition.fromEnqueuedTime(testData.getPartitionProperties().getLastEnqueuedTime());
-        final EventData expectedEvent = receivedEvents[0];
+
+        final EventPosition position = EventPosition.fromEnqueuedTime(enqueuedTime.minusMillis(1));
 
         // Act & Assert
         StepVerifier.create(consumer.receiveFromPartition(testData.getPartitionId(), position)

diff --git a/...hubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/TestUtils.java b/...hubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/TestUtils.java
@@ -6,6 +6,7 @@
 import com.azure.core.amqp.AmqpMessageConstant;
 import com.azure.core.amqp.implementation.MessageSerializer;
 import com.azure.core.util.CoreUtils;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.messaging.eventhubs.models.PartitionEvent;
 import org.apache.qpid.proton.Proton;
 import org.apache.qpid.proton.amqp.Binary;
@@ -36,7 +37,8 @@
  */
 public final class TestUtils {
     private static final MessageSerializer MESSAGE_SERIALIZER = new EventHubMessageSerializer();
-
+    private static final ClientLogger LOGGER = new ClientLogger(TestUtils.class);
+;
     // System and application properties from the generated test message.
     static final Instant ENQUEUED_TIME = Instant.ofEpochSecond(1561344661);
     static final Long OFFSET = 1534L;
@@ -161,6 +163,13 @@ public static boolean isMatchingEvent(PartitionEvent partitionEvent, String expe
      * Checks the {@link #MESSAGE_ID} to see if it matches the {@code expectedValue}.
      */
     public static boolean isMatchingEvent(EventData event, String expectedValue) {
+        LOGGER.atInfo()
+            .addKeyValue("expectedMessageId", expectedValue)
+            .addKeyValue("sequenceNo", event.getSequenceNumber())
+            .addKeyValue("enqueuedTime", event.getEnqueuedTime())
+            .addKeyValue("MESSAGE_ID", event.getProperties() == null ? null : event.getProperties().get(MESSAGE_ID))
+            .log("isMatchingEvent");
+
         return event.getProperties() != null && event.getProperties().containsKey(MESSAGE_ID)
             && expectedValue.equals(event.getProperties().get(MESSAGE_ID));
     }