SQS Autoinstrumentation not providing links

[ImDevinC]

I'm using the following configuration, called with NODE_OPTIONS=--require ./telemetry.js and the code examples below. All of my traces properly show up, but the consumer is not linked to the producer for some reason. When I look at the message payload in the consumer, I can see that the trace parent context is available as a message attribute, but when I look at my trace in Jaeger, I'm not seeing any linking between the two services. Any insight is appreciated.

telemetry.js

const {
  getNodeAutoInstrumentations,
} = require("@opentelemetry/auto-instrumentations-node");
const {
  OTLPTraceExporter,
} = require("@opentelemetry/exporter-trace-otlp-proto");
const { NodeSDK } = require("@opentelemetry/sdk-node");
const { BatchSpanProcessor } = require("@opentelemetry/sdk-trace-base");

const exporter = new OTLPTraceExporter({
  url:
    "http://" +
    (process.env.OTEL_COLLECTOR_URL ?? "localhost:4318") +
    "/v1/traces",
});

const sdk = new NodeSDK({
  spanProcessor: new BatchSpanProcessor(exporter),
  serviceName: process.env.SERVICE_NAME,
  instrumentations: [
    getNodeAutoInstrumentations({
      "@opentelemetry/instrumentation-fs": {
        enabled: false,
      },
      "@opentelemetry/instrumentation-aws-sdk": {
        sqsExtractContextPropagationFromPayload: true,
        suppressInternalInstrumentation: true,
      },
    }),
  ],
});

sdk.start();

process.on("SIGTERM", () => {
  sdk
    .shutdown()
    .then(() => console.log("Tracing terminated"))
    .catch((error) => console.log("Error terminating tracing", error))
    .finally(() => process.exit(0));
});

publish.ts

import {
  GetQueueUrlCommand,
  SQSClient,
  SendMessageCommand,
} from "@aws-sdk/client-sqs";
import { SpanStatusCode, trace } from "@opentelemetry/api";

const tracer = trace.getTracer(process.env.SERVICE_NAME!);

const start = async () => {
  return tracer.startActiveSpan("publish", async (span) => {
    try {
      await publish();
    } catch (err) {
      span.recordException(err);
      span.setStatus({
        code: SpanStatusCode.ERROR,
        message: JSON.stringify(err),
      });
    }
    span.end();
  });
};

const publish = async () => {
  const sqs = new SQSClient({});

  const getQueueUrlCommand = new GetQueueUrlCommand({
    QueueName: process.env.QUEUE_NAME,
  });

  let queueUrl = "";

  try {
    const response = await sqs.send(getQueueUrlCommand);
    if (!response.QueueUrl) {
      throw new Error("no queue url");
    }
    queueUrl = response.QueueUrl;
  } catch (err) {
    console.log(err);
    return;
  }

  const publishCommand = new SendMessageCommand({
    QueueUrl: queueUrl,
    MessageBody: "Hello World!",
  });

  try {
    const response = await sqs.send(publishCommand);
    console.log(JSON.stringify(response));
  } catch (err) {
    console.log(err);
  }
};

start().then(async () => {
  return new Promise((resolve) => {
    setTimeout(resolve, 5000);
  }).then(() => {
    console.log("done");
  });
});

consume.ts

import { PutObjectCommand, S3Client } from "@aws-sdk/client-s3";
import {
  ReceiveMessageCommand,
  Message,
  SQSClient,
  DeleteMessageCommand,
  GetQueueUrlCommand,
} from "@aws-sdk/client-sqs";
import { SpanStatusCode, trace } from "@opentelemetry/api";

const tracer = trace.getTracer(process.env.SERVICE_NAME!);

const start = async () => {
  return tracer.startActiveSpan("consume", async (span) => {
    try {
      await consume();
    } catch (err) {
      span.recordException(err);
      span.setStatus({
        code: SpanStatusCode.ERROR,
        message: JSON.stringify(err),
      });
    }
    span.end();
  });
};

const consume = async () => {
  const sqs = new SQSClient({});
  const getQueueUrlCommand = new GetQueueUrlCommand({
    QueueName: process.env.QUEUE_NAME,
  });

  let queueUrl = "";

  try {
    const response = await sqs.send(getQueueUrlCommand);
    if (!response.QueueUrl) {
      throw new Error("no queue url");
    }
    queueUrl = response.QueueUrl;
  } catch (err) {
    throw err;
  }

  const publishCommand = new ReceiveMessageCommand({
    QueueUrl: queueUrl,
  });

  let messages: Message[] = [];

  try {
    const response = await sqs.send(publishCommand);
    messages = response.Messages || [];
  } catch (err) {
    throw err;
  }

  if (messages.length == 0) {
    console.log("No messages found");
    return;
  }

  const s3 = new S3Client({});
  for (const message of messages) {
    console.log(JSON.stringify(message));
    const putCommand = new PutObjectCommand({
      Bucket: process.env.BUCKET_NAME,
      Key: "message.txt",
      Body: message.Body,
    });
    const deleteMessageCommand = new DeleteMessageCommand({
      QueueUrl: queueUrl,
      ReceiptHandle: message.ReceiptHandle,
    });
    try {
      await s3.send(putCommand);
      await sqs.send(deleteMessageCommand);
    } catch (err) {
      throw err;
    }
  }
};

start().then(async () => {
  return new Promise((resolve) => {
    setTimeout(resolve, 5000);
  }).then(() => {
    console.log("done");
  });
});

[ImDevinC]

Closing this in favor of the issue here as directed on Slack: open-telemetry/opentelemetry-js-contrib#1814