Skip to content

Commit f35de28

Browse files
kamalaboulhosnkamalaboulhosn
authored
docs(sample): Add samples for topic and subscription SMTs (#2388)
* docs(sample): Add samples for topic and subscription SMTs * docs(sample): Add samples for topic and subscription SMTs * Fix missing paren * Formatting fixes
1 parent b2a3e35 commit f35de28

File tree

3 files changed

+170
-0
lines changed

3 files changed

+170
-0
lines changed

‎samples/snippets/src/main/java/pubsub/CreateSubscriptionWithSmtExample.java‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
/*
2+
* Copyright 2025 Google LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package pubsub;
18+
19+
// [START pubsub_create_subscription_with_smt]
20+
import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
21+
import com.google.pubsub.v1.JavaScriptUDF;
22+
import com.google.pubsub.v1.MessageTransform;
23+
import com.google.pubsub.v1.ProjectSubscriptionName;
24+
import com.google.pubsub.v1.ProjectTopicName;
25+
import com.google.pubsub.v1.Subscription;
26+
import java.io.IOException;
27+
28+
public class CreateSubscriptionWithSmtExample {
29+
public static void main(String... args) throws Exception {
30+
// TODO(developer): Replace these variables before running the sample.
31+
String projectId = "your-project-id";
32+
String topicId = "your-topic-id";
33+
String subscriptionId = "your-subscription-id";
34+
35+
createSubscriptionWithSmtExample(projectId, topicId, subscriptionId);
36+
}
37+
38+
public static void createSubscriptionWithSmtExample(
39+
String projectId, String topicId, String subscriptionId) throws IOException {
40+
41+
// UDF that removes the 'ssn' field, if present
42+
String code =
43+
"function redactSSN(message, metadata) {"
44+
+ " const data = JSON.parse(message.data);"
45+
+ " delete data['ssn'];"
46+
+ " message.data = JSON.stringify(data);"
47+
+ " return message;"
48+
+ "}";
49+
String functionName = "redactSSN";
50+
51+
JavaScriptUDF udf =
52+
JavaScriptUDF.newBuilder().setCode(code).setFunctionName(functionName).build();
53+
MessageTransform transform = MessageTransform.newBuilder().setJavascriptUdf(udf).build();
54+
55+
try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {
56+
57+
ProjectTopicName topicName = ProjectTopicName.of(projectId, topicId);
58+
ProjectSubscriptionName subscriptionName =
59+
ProjectSubscriptionName.of(projectId, subscriptionId);
60+
61+
Subscription subscription =
62+
subscriptionAdminClient.createSubscription(
63+
Subscription.newBuilder()
64+
.setName(subscriptionName.toString())
65+
.setTopic(topicName.toString())
66+
// Add the UDF message transform
67+
.addMessageTransforms(transform)
68+
.build());
69+
70+
System.out.println("Created subscription with SMT: " + subscription.getAllFields());
71+
}
72+
}
73+
}
74+
// [END pubsub_create_subscription_with_smt]

‎samples/snippets/src/main/java/pubsub/CreateTopicWithSmtExample.java‎

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
/*
2+
* Copyright 2025 Google LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package pubsub;
18+
19+
// [START pubsub_create_topic_with_smt]
20+
21+
import com.google.api.gax.rpc.AlreadyExistsException;
22+
import com.google.cloud.pubsub.v1.TopicAdminClient;
23+
import com.google.pubsub.v1.JavaScriptUDF;
24+
import com.google.pubsub.v1.MessageTransform;
25+
import com.google.pubsub.v1.Topic;
26+
import com.google.pubsub.v1.TopicName;
27+
import java.io.IOException;
28+
29+
public class CreateTopicWithSmtExample {
30+
31+
public static void main(String... args) throws Exception {
32+
// TODO(developer): Replace these variables before running the sample.
33+
String projectId = "your-project-id";
34+
String topicId = "your-topic-id";
35+
36+
createTopicWithSmtExample(projectId, topicId);
37+
}
38+
39+
public static void createTopicWithSmtExample(String projectId, String topicId)
40+
throws IOException {
41+
TopicName topicName = TopicName.of(projectId, topicId);
42+
43+
// UDF that removes the 'ssn' field, if present
44+
String code =
45+
"function redactSSN(message, metadata) {"
46+
+ " const data = JSON.parse(message.data);"
47+
+ " delete data['ssn'];"
48+
+ " message.data = JSON.stringify(data);"
49+
+ " return message;"
50+
+ "}";
51+
String functionName = "redactSSN";
52+
53+
JavaScriptUDF udf =
54+
JavaScriptUDF.newBuilder().setCode(code).setFunctionName(functionName).build();
55+
MessageTransform transform = MessageTransform.newBuilder().setJavascriptUdf(udf).build();
56+
try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) {
57+
58+
Topic topic =
59+
topicAdminClient.createTopic(
60+
Topic.newBuilder()
61+
.setName(topicName.toString())
62+
// Add the UDF message transform
63+
.addMessageTransforms(transform)
64+
.build());
65+
66+
System.out.println("Created topic with SMT: " + topic.getName());
67+
} catch (AlreadyExistsException e) {
68+
System.out.println(topicName + "already exists.");
69+
}
70+
}
71+
}
72+
// [END pubsub_create_topic_with_smt]

‎samples/snippets/src/test/java/pubsub/AdminIT.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -60,10 +60,12 @@ public class AdminIT {
6060
"confluent-cloud-ingestion-topic-" + _suffix;
6161
private static final String azureEventHubsIngestionTopicId =
6262
"azure-event-hubs-ingestion-topic-" + _suffix;
63+
private static final String smtTopicId = "smt-topic-" + _suffix;
6364
private static final String pullSubscriptionId = "iam-pull-subscription-" + _suffix;
6465
private static final String pushSubscriptionId = "iam-push-subscription-" + _suffix;
6566
private static final String orderedSubscriptionId = "iam-ordered-subscription-" + _suffix;
6667
private static final String filteredSubscriptionId = "iam-filtered-subscription-" + _suffix;
68+
private static final String smtSubscriptionId = "smt-subscription-" + _suffix;
6769
private static final String exactlyOnceSubscriptionId =
6870
"iam-exactly-once-subscription-" + _suffix;
6971
private static final String pushEndpoint = "https://my-test-project.appspot.com/push";
@@ -118,6 +120,8 @@ public class AdminIT {
118120
TopicName.of(projectId, confluentCloudIngestionTopicId);
119121
private static final TopicName azureEventHubsIngestionTopicName =
120122
TopicName.of(projectId, azureEventHubsIngestionTopicId);
123+
private static final TopicName smtTopicName =
124+
TopicName.of(projectId, smtTopicId);
121125
private static final SubscriptionName pullSubscriptionName =
122126
SubscriptionName.of(projectId, pullSubscriptionId);
123127
private static final SubscriptionName pushSubscriptionName =
@@ -128,6 +132,8 @@ public class AdminIT {
128132
SubscriptionName.of(projectId, filteredSubscriptionId);
129133
private static final SubscriptionName exactlyOnceSubscriptionName =
130134
SubscriptionName.of(projectId, exactlyOnceSubscriptionId);
135+
private static final SubscriptionName smtSubscriptionName =
136+
SubscriptionName.of(projectId, smtSubscriptionId);
131137

132138
private static void requireEnvVar(String varName) {
133139
assertNotNull(
@@ -456,5 +462,23 @@ public void testAdmin() throws Exception {
456462
// Test delete Azure Event Hubs ingestion topic.
457463
DeleteTopicExample.deleteTopicExample(projectId, azureEventHubsIngestionTopicId);
458464
assertThat(bout.toString()).contains("Deleted topic.");
465+
466+
bout.reset();
467+
// Test create topic with an SMT.
468+
CreateTopicWithSmtExample.createTopicWithSmtExample(
469+
projectId, smtTopicId);
470+
assertThat(bout.toString())
471+
.contains("Created topic with SMT: " + smtTopicName.toString());
472+
473+
bout.reset();
474+
// Test create topic with an SMT.
475+
CreateSubscriptionWithSmtExample.createSubscriptionWithSmtExample(
476+
projectId, smtTopicId, smtSubscriptionId);
477+
assertThat(bout.toString())
478+
.contains("Created subscription with SMT");
479+
assertThat(bout.toString()).contains(smtSubscriptionName.toString());
480+
assertThat(bout.toString()).contains("redactSSN");
481+
DeleteSubscriptionExample.deleteSubscriptionExample(projectId, smtSubscriptionId);
482+
DeleteTopicExample.deleteTopicExample(projectId, smtTopicId);
459483
}
460484
}

0 commit comments

Comments
 (0)