Kafka模板连接头（CorrelationId）未发送到Google Pub Sub。

要解决Kafka模板连接头（CorrelationId）未发送到Google Pub Sub的问题，你可以按照以下步骤进行操作：

1. 确保在Kafka模板中正确设置了连接头（CorrelationId）。你可以使用KafkaTemplate的send()方法发送消息时，为ProducerRecord对象设置连接头。示例如下：

ProducerRecord<String, String> record = new ProducerRecord<>("topicName", "key", "value");
record.headers().add("CorrelationId", "your-correlation-id".getBytes());
kafkaTemplate.send(record);

2. 确保你的Google Pub Sub消费者正确接收并处理了连接头。你需要在Google Pub Sub消费者的订阅中添加对连接头的处理逻辑。示例如下：

import com.google.cloud.pubsub.v1.MessageReceiver;
import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.pubsub.v1.PubsubMessage;

public class MyMessageReceiver implements MessageReceiver {
    @Override
    public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
        // 获取连接头（CorrelationId）
        String correlationId = message.getAttributesOrDefault("CorrelationId", "");
        
        // 处理消息逻辑
        
        // 手动确认消息
        consumer.ack();
    }
}

3. 确保你的Google Pub Sub消费者订阅正确配置了连接头。你需要在订阅的消息流中设置连接头。示例如下：

import com.google.pubsub.v1.SubscriptionName;
import com.google.cloud.pubsub.v1.Subscriber;
import com.google.cloud.pubsub.v1.Subscriber.Builder;

String projectId = "your-project-id";
String subscriptionId = "your-subscription-id";
String topicId = "your-topic-id";

SubscriptionName subscriptionName = SubscriptionName.of(projectId, subscriptionId);
Builder subscriberBuilder = Subscriber.newBuilder(subscriptionName, new MyMessageReceiver());

// 设置连接头（CorrelationId）
subscriberBuilder.setReceiverOptions(SubscriberReceiverOptions.newBuilder()
    .setReceiverLabels(ImmutableMap.of("CorrelationId", "your-correlation-id"))
    .build());

Subscriber subscriber = subscriberBuilder.build();
subscriber.startAsync().awaitRunning();

通过以上步骤，你应该能够在Kafka模板中正确发送连接头（CorrelationId）到Google Pub Sub，并在消费者端正确接收和处理连接头。