Merge branch 'main' into fix/flush-before-requeuing-retries

1kilobit · web-flow · commit fe99d8a0ff29 · 2025-05-16T09:15:51.000-07:00
diff --git a/.github/.OwlBot.lock.yaml b/.github/.OwlBot.lock.yaml
@@ -14,3 +14,4 @@
 docker:
   image: gcr.io/cloud-devrel-public-resources/owlbot-nodejs:latest
   digest: sha256:c7e4968cfc97a204a4b2381f3ecb55cabc40c4cccf88b1ef8bef0d976be87fee
+# created: 2025-04-08T17:33:08.498793944Z
diff --git a/.github/scripts/package.json b/.github/scripts/package.json
@@ -14,7 +14,7 @@
         "js-yaml": "^4.1.0"
     },
     "devDependencies": {
-        "@octokit/rest": "^19.0.0",
+        "@octokit/rest": "^21.0.0",
         "mocha": "^10.0.0",
         "sinon": "^18.0.0"
     }
diff --git a/package.json b/package.json
@@ -54,7 +54,7 @@
     "@google-cloud/promisify": "^5.0.0",
     "@opentelemetry/api": "~1.9.0",
     "@opentelemetry/core": "^1.30.1",
-    "@opentelemetry/semantic-conventions": "~1.32.0",
+    "@opentelemetry/semantic-conventions": "~1.33.0",
     "arrify": "^2.0.0",
     "extend": "^3.0.2",
     "google-auth-library": "^10.0.0-rc.1",
@@ -80,7 +80,7 @@
     "@types/uuid": "^10.0.0",
     "c8": "^10.1.3",
     "codecov": "^3.8.3",
-    "execa": "~5.0.0",
+    "execa": "~5.1.0",
     "gapic-tools": "^1.0.1",
     "gts": "^6.0.2",
     "jsdoc": "^4.0.4",
diff --git a/samples/package.json b/samples/package.json
@@ -34,14 +34,14 @@
     "@opentelemetry/semantic-conventions": "^1.17.0",
     "avro-js": "^1.11.3",
     "p-defer": "^3.0.0",
-    "protobufjs": "~7.4.0"
+    "protobufjs": "~7.5.0"
   },
   "devDependencies": {
-    "@google-cloud/bigquery": "^7.0.0",
+    "@google-cloud/bigquery": "^8.0.0",
     "@google-cloud/typeless-sample-bot": "^3.0.0",
     "@types/chai": "^4.2.16",
     "chai": "^4.2.0",
-    "gts": "^5.0.0",
+    "gts": "^6.0.0",
     "lru-cache": "9.1.2",
     "mocha": "^9.2.2",
     "rimraf": "5.0.9",
diff --git a/samples/publishMessage.js b/samples/publishMessage.js
@@ -51,7 +51,7 @@ async function publishMessage(topicNameOrId, data) {
   const topic = pubSubClient.topic(topicNameOrId);
 
   try {
-    const messageId = topic.publishMessage({data: dataBuffer});
+    const messageId = await topic.publishMessage({data: dataBuffer});
     console.log(`Message ${messageId} published.`);
   } catch (error) {
     console.error(`Received error while publishing: ${error.message}`);
diff --git a/samples/publishMessageWithCustomAttributes.js b/samples/publishMessageWithCustomAttributes.js
@@ -55,7 +55,7 @@ async function publishMessageWithCustomAttributes(topicNameOrId, data) {
   // Cache topic objects (publishers) and reuse them.
   const topic = pubSubClient.topic(topicNameOrId);
 
-  const messageId = topic.publishMessage({
+  const messageId = await topic.publishMessage({
     data: dataBuffer,
     attributes: customAttributes,
   });
diff --git a/samples/publishOrderedMessage.js b/samples/publishOrderedMessage.js
@@ -72,7 +72,7 @@ async function publishOrderedMessage(topicNameOrId, data, orderingKey) {
   const topic = pubSubClient.topic(topicNameOrId, publishOptions);
 
   // Publishes the message
-  const messageId = topic.publishMessage(message);
+  const messageId = await topic.publishMessage(message);
 
   console.log(`Message ${messageId} published.`);
 
diff --git a/samples/system-test/schema.test.ts b/samples/system-test/schema.test.ts
@@ -383,7 +383,7 @@ describe('schema', () => {
     const topic = await createTopicWithSchema(id, schema.id, Encodings.Json);
     const sub = await createSub(id, topic.name);
 
-    topic.publishMessage({
+    await topic.publishMessage({
       data: Buffer.from(
         JSON.stringify({
           name: 'Alberta',
@@ -419,7 +419,7 @@ describe('schema', () => {
     );
     const sub = await createSub(id, topic.name);
 
-    topic.publishMessage({
+    await topic.publishMessage({
       data: Buffer.from(
         JSON.stringify({
           name: 'Alberta',
@@ -443,7 +443,7 @@ describe('schema', () => {
     const topic = await createTopicWithSchema(id, schema.id, Encodings.Json);
     const sub = await createSub(id, topic.name);
 
-    topic.publishMessage({
+    await topic.publishMessage({
       data: Buffer.from(
         JSON.stringify({
           name: 'Quebec',
diff --git a/samples/typescript/publishMessage.ts b/samples/typescript/publishMessage.ts
@@ -47,7 +47,7 @@ async function publishMessage(topicNameOrId: string, data: string) {
   const topic = pubSubClient.topic(topicNameOrId);
 
   try {
-    const messageId = topic.publishMessage({data: dataBuffer});
+    const messageId = await topic.publishMessage({data: dataBuffer});
     console.log(`Message ${messageId} published.`);
   } catch (error) {
     console.error(
diff --git a/samples/typescript/publishMessageWithCustomAttributes.ts b/samples/typescript/publishMessageWithCustomAttributes.ts
@@ -58,7 +58,7 @@ async function publishMessageWithCustomAttributes(
   // Cache topic objects (publishers) and reuse them.
   const topic = pubSubClient.topic(topicNameOrId);
 
-  const messageId = topic.publishMessage({
+  const messageId = await topic.publishMessage({
     data: dataBuffer,
     attributes: customAttributes,
   });
diff --git a/samples/typescript/publishOrderedMessage.ts b/samples/typescript/publishOrderedMessage.ts
@@ -72,7 +72,7 @@ async function publishOrderedMessage(
   const topic = pubSubClient.topic(topicNameOrId, publishOptions);
 
   // Publishes the message
-  const messageId = topic.publishMessage(message);
+  const messageId = await topic.publishMessage(message);
 
   console.log(`Message ${messageId} published.`);
 
diff --git a/src/telemetry-tracing.ts b/src/telemetry-tracing.ts
@@ -305,6 +305,7 @@ export class PubsubSpans {
     params: AttributeParams,
     message?: PubsubMessage,
     caller?: string,
+    operation?: string,
   ): SpanAttributes {
     const destinationName = params.topicName ?? params.subName;
     const destinationId = params.topicId ?? params.subId;
@@ -351,6 +352,9 @@ export class PubsubSpans {
       if (message.ackId) {
         spanAttributes['messaging.gcp_pubsub.message.ack_id'] = message.ackId;
       }
+      if (operation) {
+        spanAttributes['messaging.operation'] = operation;
+      }
     }
 
     return spanAttributes;
@@ -368,11 +372,15 @@ export class PubsubSpans {
     const topicInfo = getTopicInfo(topicName);
     const span: Span = getTracer().startSpan(`${topicName} create`, {
       kind: SpanKind.PRODUCER,
-      attributes: PubsubSpans.createAttributes(topicInfo, message, caller),
+      attributes: PubsubSpans.createAttributes(
+        topicInfo,
+        message,
+        caller,
+        'create',
+      ),
     });
     if (topicInfo.topicId) {
       span.updateName(`${topicInfo.topicId} create`);
-      span.setAttribute('messaging.operation', 'create');
       span.setAttribute('messaging.destination.name', topicInfo.topicId);
     }
 
@@ -404,10 +412,14 @@ export class PubsubSpans {
 
     const subInfo = getSubscriptionInfo(subName);
     const name = `${subInfo.subId ?? subName} subscribe`;
-    const attributes = this.createAttributes(subInfo, message, caller);
+    const attributes = this.createAttributes(
+      subInfo,
+      message,
+      caller,
+      'receive',
+    );
     if (subInfo.subId) {
       attributes['messaging.destination.name'] = subInfo.subId;
-      attributes['messaging.operation'] = 'receive';
     }
 
     if (context) {
@@ -473,6 +485,7 @@ export class PubsubSpans {
       getTopicInfo(topicName),
       undefined,
       caller,
+      'create',
     );
     const links: Link[] = messages
       .filter(m => m.parentSpan && isSampled(m.parentSpan))
@@ -515,6 +528,7 @@ export class PubsubSpans {
       subInfo,
       undefined,
       caller,
+      'receive',
     );
     const links: Link[] = messageSpans
       .filter(m => m && isSampled(m))
@@ -531,7 +545,6 @@ export class PubsubSpans {
     );
 
     span?.setAttribute('messaging.batch.message_count', messageSpans.length);
-    span?.setAttribute('messaging.operation', 'receive');
 
     if (span) {
       // Also attempt to link from the subscribe span(s) back to the publish RPC span.
@@ -563,6 +576,7 @@ export class PubsubSpans {
       subInfo,
       undefined,
       caller,
+      'receive',
     );
     const links: Link[] = messageSpans
       .filter(m => m && isSampled(m))
@@ -579,7 +593,6 @@ export class PubsubSpans {
     );
 
     span?.setAttribute('messaging.batch.message_count', messageSpans.length);
-    span?.setAttribute('messaging.operation', 'receive');
 
     if (span) {
       // Also attempt to link from the subscribe span(s) back to the publish RPC span.
diff --git a/system-test/fixtures/sample/package.json b/system-test/fixtures/sample/package.json
@@ -18,6 +18,6 @@
   "devDependencies": {
     "@types/node": "^22.0.0",
     "typescript": "^5.1.6",
-    "gts": "^5.0.0"
+    "gts": "^6.0.0"
   }
 }
diff --git a/test/telemetry-tracing.ts b/test/telemetry-tracing.ts
@@ -212,6 +212,7 @@ describe('OpenTelemetryTracer', () => {
         topicInfo,
         message,
         'tests',
+        'create',
       );
       assert.deepStrictEqual(topicAttrs, {
         'messaging.system': 'gcp_pubsub',
@@ -222,6 +223,7 @@ describe('OpenTelemetryTracer', () => {
         'messaging.gcp_pubsub.message.exactly_once_delivery':
           message.isExactlyOnceDelivery,
         'messaging.gcp_pubsub.message.ack_id': message.ackId,
+        'messaging.operation': 'create',
         'code.function': 'tests',
       });
 
@@ -235,10 +237,12 @@ describe('OpenTelemetryTracer', () => {
         topicInfo,
         message,
         'tests',
+        'create',
       );
       assert.deepStrictEqual(topicAttrs2, {
         'messaging.system': 'gcp_pubsub',
         'messaging.destination.name': topicInfo.topicId,
+        'messaging.operation': 'create',
         'gcp.project_id': topicInfo.projectId,
         'messaging.message.envelope.size': message.data?.length,
         'code.function': 'tests',
@@ -283,6 +287,7 @@ describe('OpenTelemetryTracer', () => {
       const firstSpan = spans.pop();
       assert.ok(firstSpan);
       assert.strictEqual(firstSpan.name, `${tests.topicInfo.topicId} create`);
+      assert.strictEqual(firstSpan.attributes['messaging.operation'], 'create');
       assert.strictEqual(
         firstSpan.attributes['messaging.destination.name'],
         tests.topicInfo.topicId,
@@ -312,7 +317,6 @@ describe('OpenTelemetryTracer', () => {
       const firstSpan = spans.pop();
       assert.ok(firstSpan);
       assert.strictEqual(firstSpan.name, 'other create');
-
       assert.strictEqual(
         firstSpan.attributes['messaging.destination.name'],
         'other',
@@ -346,6 +350,10 @@ describe('OpenTelemetryTracer', () => {
         childReadSpan.attributes['messaging.destination.name'],
         'sub',
       );
+      assert.strictEqual(
+        childReadSpan.attributes['messaging.operation'],
+        'receive',
+      );
       assert.strictEqual(childReadSpan.kind, SpanKind.CONSUMER);
       assert.ok(childReadSpan.parentSpanId);
     });

Original file line number	Diff line number	Diff line change
`@@ -14,7 +14,7 @@`
`14`	`14`	`"js-yaml": "^4.1.0"`
`15`	`15`	`},`
`16`	`16`	`"devDependencies": {`
`17`		`- "@octokit/rest": "^19.0.0",`
	`17`	`+ "@octokit/rest": "^21.0.0",`
`18`	`18`	`"mocha": "^10.0.0",`
`19`	`19`	`"sinon": "^18.0.0"`
`20`	`20`	`}`