46
46
import com .google .cloud .pubsub .v1 .stub .PublisherStub ;
47
47
import com .google .cloud .pubsub .v1 .stub .PublisherStubSettings ;
48
48
import com .google .common .base .Preconditions ;
49
+ import com .google .protobuf .CodedOutputStream ;
49
50
import com .google .pubsub .v1 .PublishRequest ;
50
51
import com .google .pubsub .v1 .PublishResponse ;
51
52
import com .google .pubsub .v1 .PubsubMessage ;
@@ -99,6 +100,7 @@ public class Publisher implements PublisherInterface {
99
100
private static final String OPEN_TELEMETRY_TRACER_NAME = "com.google.cloud.pubsub.v1" ;
100
101
101
102
private final String topicName ;
103
+ private final int topicNameSize ;
102
104
103
105
private final BatchingSettings batchingSettings ;
104
106
private final boolean enableMessageOrdering ;
@@ -145,6 +147,8 @@ public static long getApiMaxRequestBytes() {
145
147
146
148
private Publisher (Builder builder ) throws IOException {
147
149
topicName = builder .topicName ;
150
+ topicNameSize =
151
+ CodedOutputStream .computeStringSize (PublishRequest .TOPIC_FIELD_NUMBER , this .topicName );
148
152
149
153
this .batchingSettings = builder .batchingSettings ;
150
154
FlowControlSettings flowControl = this .batchingSettings .getFlowControlSettings ();
@@ -309,7 +313,7 @@ public ApiFuture<String> publish(PubsubMessage message) {
309
313
}
310
314
MessagesBatch messagesBatch = messagesBatches .get (orderingKey );
311
315
if (messagesBatch == null ) {
312
- messagesBatch = new MessagesBatch (batchingSettings , orderingKey );
316
+ messagesBatch = new MessagesBatch (batchingSettings , topicNameSize , orderingKey );
313
317
messagesBatches .put (orderingKey , messagesBatch );
314
318
}
315
319
@@ -636,7 +640,9 @@ private static final class OutstandingPublish {
636
640
OutstandingPublish (PubsubMessageWrapper messageWrapper ) {
637
641
this .publishResult = SettableApiFuture .create ();
638
642
this .messageWrapper = messageWrapper ;
639
- this .messageSize = messageWrapper .getPubsubMessage ().getSerializedSize ();
643
+ this .messageSize =
644
+ CodedOutputStream .computeMessageSize (
645
+ PublishRequest .MESSAGES_FIELD_NUMBER , messageWrapper .getPubsubMessage ());
640
646
}
641
647
}
642
648
@@ -1093,12 +1099,15 @@ void release(long messageSize) {
1093
1099
1094
1100
private class MessagesBatch {
1095
1101
private List <OutstandingPublish > messages ;
1102
+ private int initialBatchedBytes ;
1096
1103
private int batchedBytes ;
1097
1104
private String orderingKey ;
1098
1105
private final BatchingSettings batchingSettings ;
1099
1106
1100
- private MessagesBatch (BatchingSettings batchingSettings , String orderingKey ) {
1107
+ private MessagesBatch (
1108
+ BatchingSettings batchingSettings , int initialBatchedBytes , String orderingKey ) {
1101
1109
this .batchingSettings = batchingSettings ;
1110
+ this .initialBatchedBytes = initialBatchedBytes ;
1102
1111
this .orderingKey = orderingKey ;
1103
1112
reset ();
1104
1113
}
@@ -1111,7 +1120,7 @@ private OutstandingBatch popOutstandingBatch() {
1111
1120
1112
1121
private void reset () {
1113
1122
messages = new LinkedList <>();
1114
- batchedBytes = 0 ;
1123
+ batchedBytes = initialBatchedBytes ;
1115
1124
}
1116
1125
1117
1126
private boolean isEmpty () {
@@ -1150,7 +1159,9 @@ && getBatchedBytes() + outstandingPublish.messageSize >= getMaxBatchBytes()) {
1150
1159
// immediately.
1151
1160
// Alternatively if after adding the message we have reached the batch max messages then we
1152
1161
// have a batch to send.
1153
- if ((hasBatchingBytes () && outstandingPublish .messageSize >= getMaxBatchBytes ())
1162
+ // Note that exceeding {@link Publisher#getApiMaxRequestBytes()} will result in failed
1163
+ // publishes without compression and may yet fail if a request is not sufficiently compressed.
1164
+ if ((hasBatchingBytes () && getBatchedBytes () >= getMaxBatchBytes ())
1154
1165
|| getMessagesCount () == batchingSettings .getElementCountThreshold ()) {
1155
1166
batchesToSend .add (popOutstandingBatch ());
1156
1167
}
0 commit comments