Merge branch 'master' into pubsub-ordering-keys

sduskis · sduskis · commit ad54d6980fb0 · 2019-04-15T11:31:06.000-04:00
diff --git a/google-api-grpc/proto-google-cloud-pubsub-v1/src/main/java/com/google/pubsub/v1/PubsubMessage.java b/google-api-grpc/proto-google-cloud-pubsub-v1/src/main/java/com/google/pubsub/v1/PubsubMessage.java
diff --git a/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java b/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java
@@ -220,7 +220,7 @@ public ApiFuture<String> publish(PubsubMessage message) {
     final String orderingKey = message.getOrderingKey();
     if (orderingKey != null && !orderingKey.isEmpty() && !enableMessageOrdering) {
       throw new IllegalStateException(
-          "Cannot publish a message with an ordering key when message ordeirng is not enabled.");
+          "Cannot publish a message with an ordering key when message ordering is not enabled.");
     }
 
     message = messageTransform.apply(message);
diff --git a/google-cloud-clients/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/MessageDispatcherTest.java b/google-cloud-clients/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/MessageDispatcherTest.java
@@ -28,9 +28,7 @@
 import com.google.pubsub.v1.ReceivedMessage;
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -44,10 +42,16 @@ public class MessageDispatcherTest {
           .setAckId("ackid")
           .setMessage(PubsubMessage.newBuilder().setData(ByteString.EMPTY).build())
           .build();
+  private static final Runnable NOOP_RUNNABLE =
+      new Runnable() {
+        @Override
+        public void run() {
+          // No-op; don't do anything.
+        }
+      };
 
   private MessageDispatcher dispatcher;
   private LinkedBlockingQueue<AckReplyConsumer> consumers;
-  private Map<String, List<ByteString>> messagesByOrderingKey;
   private List<String> sentAcks;
   private List<ModAckItem> sentModAcks;
   private FakeClock clock;
@@ -67,20 +71,13 @@ static ModAckItem of(String ackId, int seconds) {
   @Before
   public void setUp() {
     consumers = new LinkedBlockingQueue<>();
-    messagesByOrderingKey = new HashMap<>();
     sentAcks = new ArrayList<>();
     sentModAcks = new ArrayList<>();
 
     MessageReceiver receiver =
         new MessageReceiver() {
           @Override
           public void receiveMessage(final PubsubMessage message, final AckReplyConsumer consumer) {
-            List<ByteString> messages = messagesByOrderingKey.get(message.getOrderingKey());
-            if (messages == null) {
-              messages = new ArrayList<>();
-              messagesByOrderingKey.put(message.getOrderingKey(), messages);
-            }
-            messages.add(message.getData());
             consumers.add(consumer);
           }
         };
@@ -206,47 +203,4 @@ public void testDeadlineAdjustment() throws Exception {
 
     assertThat(dispatcher.computeDeadlineSeconds()).isEqualTo(42);
   }
-
-  private ReceivedMessage newReceivedMessage(String ackId, String orderingKey, String data) {
-    return ReceivedMessage.newBuilder()
-        .setAckId(ackId)
-        .setMessage(
-            PubsubMessage.newBuilder()
-                .setOrderingKey(orderingKey)
-                .setData(ByteString.copyFromUtf8(data))
-                .build())
-        .build();
-  }
-
-  @Test
-  public void testOrderingKey() throws Exception {
-    // Create messages with "orderA".
-    ReceivedMessage message1 = newReceivedMessage("ackId1", "orderA", "m1");
-    ReceivedMessage message2 = newReceivedMessage("ackId2", "orderA", "m2");
-    // Create messages with "orderB".
-    ReceivedMessage message3 = newReceivedMessage("ackId3", "orderB", "m3");
-    ReceivedMessage message4 = newReceivedMessage("ackId4", "orderB", "m4");
-    ReceivedMessage message5 = newReceivedMessage("ackId5", "orderB", "m5");
-
-    dispatcher.processReceivedMessages(Collections.singletonList(message1));
-    consumers.take().ack();
-    dispatcher.processReceivedMessages(Collections.singletonList(message2));
-    consumers.take().ack();
-    dispatcher.processReceivedMessages(Collections.singletonList(message3));
-    consumers.take().ack();
-    dispatcher.processReceivedMessages(Collections.singletonList(message4));
-    consumers.take().ack();
-    dispatcher.processReceivedMessages(Collections.singletonList(message5));
-    consumers.take().ack();
-
-    assertThat(messagesByOrderingKey.get("orderA"))
-        .containsExactly(ByteString.copyFromUtf8("m1"), ByteString.copyFromUtf8("m2"))
-        .inOrder();
-    assertThat(messagesByOrderingKey.get("orderB"))
-        .containsExactly(
-            ByteString.copyFromUtf8("m3"),
-            ByteString.copyFromUtf8("m4"),
-            ByteString.copyFromUtf8("m5"))
-        .inOrder();
-  }
 }

Original file line number	Diff line number	Diff line change
`@@ -220,7 +220,7 @@ public ApiFuture<String> publish(PubsubMessage message) {`
`220`	`220`	`final String orderingKey = message.getOrderingKey();`
`221`	`221`	`if (orderingKey != null && !orderingKey.isEmpty() && !enableMessageOrdering) {`
`222`	`222`	`throw new IllegalStateException(`
`223`		`- "Cannot publish a message with an ordering key when message ordeirng is not enabled.");`
	`223`	`+ "Cannot publish a message with an ordering key when message ordering is not enabled.");`
`224`	`224`	`}`
`225`	`225`
`226`	`226`	`message = messageTransform.apply(message);`