Skip to content

Commit cb4d2ac

Browse files
adinauerclaude
adinauer
and
claude
committed
ref(samples): Extract Kafka console showcase into dedicated class
Move Kafka producer/consumer showcase logic out of Main into KafkaShowcase to make the sample easier to read and follow. Keep runtime behavior unchanged by preserving the same demo entry point and flow. Co-Authored-By: Claude <noreply@anthropic.com>
1 parent 82cfc37 commit cb4d2ac

File tree

2 files changed

+108
-97
lines changed

2 files changed

+108
-97
lines changed

‎sentry-samples/sentry-samples-console/src/main/java/io/sentry/samples/console/KafkaShowcase.java‎

Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,107 @@
1+
package io.sentry.samples.console;
2+
3+
import io.sentry.ISentryLifecycleToken;
4+
import io.sentry.ITransaction;
5+
import io.sentry.Sentry;
6+
import io.sentry.kafka.SentryKafkaConsumerInterceptor;
7+
import io.sentry.kafka.SentryKafkaProducerInterceptor;
8+
import java.time.Duration;
9+
import java.util.Collections;
10+
import java.util.Properties;
11+
import java.util.UUID;
12+
import java.util.concurrent.CountDownLatch;
13+
import java.util.concurrent.TimeUnit;
14+
import org.apache.kafka.clients.consumer.ConsumerConfig;
15+
import org.apache.kafka.clients.consumer.ConsumerRecords;
16+
import org.apache.kafka.clients.consumer.KafkaConsumer;
17+
import org.apache.kafka.clients.producer.KafkaProducer;
18+
import org.apache.kafka.clients.producer.ProducerConfig;
19+
import org.apache.kafka.clients.producer.ProducerRecord;
20+
import org.apache.kafka.common.serialization.StringDeserializer;
21+
import org.apache.kafka.common.serialization.StringSerializer;
22+
23+
final class KafkaShowcase {
24+
25+
private KafkaShowcase() {}
26+
27+
static void demonstrate() {
28+
final String topic = "sentry-topic-console-sample";
29+
final CountDownLatch consumedLatch = new CountDownLatch(1);
30+
final Thread consumerThread = startKafkaConsumerThread(topic, consumedLatch);
31+
32+
final Properties producerProperties = new Properties();
33+
producerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
34+
producerProperties.put(
35+
ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
36+
producerProperties.put(
37+
ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
38+
producerProperties.put(
39+
ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, SentryKafkaProducerInterceptor.class.getName());
40+
41+
final ITransaction transaction = Sentry.startTransaction("kafka-demo", "demo");
42+
try (ISentryLifecycleToken ignored = transaction.makeCurrent()) {
43+
try (KafkaProducer<String, String> producer = new KafkaProducer<>(producerProperties)) {
44+
Thread.sleep(500);
45+
producer.send(new ProducerRecord<>(topic, "sentry-kafka sample message")).get();
46+
} catch (InterruptedException e) {
47+
Thread.currentThread().interrupt();
48+
} catch (Exception ignoredException) {
49+
// local broker may not be available when running the sample
50+
}
51+
52+
try {
53+
consumedLatch.await(5, TimeUnit.SECONDS);
54+
} catch (InterruptedException e) {
55+
Thread.currentThread().interrupt();
56+
}
57+
} finally {
58+
consumerThread.interrupt();
59+
try {
60+
consumerThread.join(1000);
61+
} catch (InterruptedException e) {
62+
Thread.currentThread().interrupt();
63+
}
64+
transaction.finish();
65+
}
66+
}
67+
68+
private static Thread startKafkaConsumerThread(
69+
final String topic, final CountDownLatch consumedLatch) {
70+
final Thread consumerThread =
71+
new Thread(
72+
() -> {
73+
final Properties consumerProperties = new Properties();
74+
consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
75+
consumerProperties.put(
76+
ConsumerConfig.GROUP_ID_CONFIG, "sentry-console-sample-" + UUID.randomUUID());
77+
consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
78+
consumerProperties.put(
79+
ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
80+
consumerProperties.put(
81+
ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
82+
StringDeserializer.class.getName());
83+
consumerProperties.put(
84+
ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
85+
SentryKafkaConsumerInterceptor.class.getName());
86+
87+
try (KafkaConsumer<String, String> consumer =
88+
new KafkaConsumer<>(consumerProperties)) {
89+
consumer.subscribe(Collections.singletonList(topic));
90+
91+
while (!Thread.currentThread().isInterrupted() && consumedLatch.getCount() > 0) {
92+
final ConsumerRecords<String, String> records =
93+
consumer.poll(Duration.ofMillis(500));
94+
if (!records.isEmpty()) {
95+
consumedLatch.countDown();
96+
break;
97+
}
98+
}
99+
} catch (Exception ignored) {
100+
// local broker may not be available when running the sample
101+
}
102+
},
103+
"sentry-kafka-sample-consumer");
104+
consumerThread.start();
105+
return consumerThread;
106+
}
107+
}

‎sentry-samples/sentry-samples-console/src/main/java/io/sentry/samples/console/Main.java‎

Lines changed: 1 addition & 97 deletions
Original file line numberDiff line numberDiff line change
@@ -3,28 +3,13 @@
33
import io.sentry.*;
44
import io.sentry.clientreport.DiscardReason;
55
import io.sentry.jcache.SentryJCacheWrapper;
6-
import io.sentry.kafka.SentryKafkaConsumerInterceptor;
7-
import io.sentry.kafka.SentryKafkaProducerInterceptor;
86
import io.sentry.protocol.Message;
97
import io.sentry.protocol.User;
10-
import java.time.Duration;
118
import java.util.Collections;
12-
import java.util.Properties;
13-
import java.util.UUID;
14-
import java.util.concurrent.CountDownLatch;
15-
import java.util.concurrent.TimeUnit;
169
import javax.cache.Cache;
1710
import javax.cache.CacheManager;
1811
import javax.cache.Caching;
1912
import javax.cache.configuration.MutableConfiguration;
20-
import org.apache.kafka.clients.consumer.ConsumerConfig;
21-
import org.apache.kafka.clients.consumer.ConsumerRecords;
22-
import org.apache.kafka.clients.consumer.KafkaConsumer;
23-
import org.apache.kafka.clients.producer.KafkaProducer;
24-
import org.apache.kafka.clients.producer.ProducerConfig;
25-
import org.apache.kafka.clients.producer.ProducerRecord;
26-
import org.apache.kafka.common.serialization.StringDeserializer;
27-
import org.apache.kafka.common.serialization.StringSerializer;
2813

2914
public class Main {
3015

@@ -198,7 +183,7 @@ public static void main(String[] args) throws InterruptedException {
198183
//
199184
// This uses the native producer interceptor from sentry-kafka.
200185
// If no local Kafka broker is available, this block exits quietly.
201-
demonstrateKafkaTracing();
186+
KafkaShowcase.demonstrate();
202187

203188
// Performance feature
204189
//
@@ -269,87 +254,6 @@ private static void captureMetrics() {
269254
Sentry.metrics().distribution("distributionMetric", 7.0);
270255
}
271256

272-
private static void demonstrateKafkaTracing() {
273-
final String topic = "sentry-topic-console-sample";
274-
final CountDownLatch consumedLatch = new CountDownLatch(1);
275-
final Thread consumerThread = startKafkaConsumerThread(topic, consumedLatch);
276-
277-
final Properties producerProperties = new Properties();
278-
producerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
279-
producerProperties.put(
280-
ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
281-
producerProperties.put(
282-
ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
283-
producerProperties.put(
284-
ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, SentryKafkaProducerInterceptor.class.getName());
285-
286-
final ITransaction transaction = Sentry.startTransaction("kafka-demo", "demo");
287-
try (ISentryLifecycleToken ignored = transaction.makeCurrent()) {
288-
try (KafkaProducer<String, String> producer = new KafkaProducer<>(producerProperties)) {
289-
Thread.sleep(500);
290-
producer.send(new ProducerRecord<>(topic, "sentry-kafka sample message")).get();
291-
} catch (InterruptedException e) {
292-
Thread.currentThread().interrupt();
293-
} catch (Exception ignoredException) {
294-
// local broker may not be available when running the sample
295-
}
296-
297-
try {
298-
consumedLatch.await(5, TimeUnit.SECONDS);
299-
} catch (InterruptedException e) {
300-
Thread.currentThread().interrupt();
301-
}
302-
} finally {
303-
consumerThread.interrupt();
304-
try {
305-
consumerThread.join(1000);
306-
} catch (InterruptedException e) {
307-
Thread.currentThread().interrupt();
308-
}
309-
transaction.finish();
310-
}
311-
}
312-
313-
private static Thread startKafkaConsumerThread(
314-
final String topic, final CountDownLatch consumedLatch) {
315-
final Thread consumerThread =
316-
new Thread(
317-
() -> {
318-
final Properties consumerProperties = new Properties();
319-
consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
320-
consumerProperties.put(
321-
ConsumerConfig.GROUP_ID_CONFIG, "sentry-console-sample-" + UUID.randomUUID());
322-
consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
323-
consumerProperties.put(
324-
ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
325-
consumerProperties.put(
326-
ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
327-
StringDeserializer.class.getName());
328-
consumerProperties.put(
329-
ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
330-
SentryKafkaConsumerInterceptor.class.getName());
331-
332-
try (KafkaConsumer<String, String> consumer =
333-
new KafkaConsumer<>(consumerProperties)) {
334-
consumer.subscribe(Collections.singletonList(topic));
335-
336-
while (!Thread.currentThread().isInterrupted() && consumedLatch.getCount() > 0) {
337-
final ConsumerRecords<String, String> records =
338-
consumer.poll(Duration.ofMillis(500));
339-
if (!records.isEmpty()) {
340-
consumedLatch.countDown();
341-
break;
342-
}
343-
}
344-
} catch (Exception ignored) {
345-
// local broker may not be available when running the sample
346-
}
347-
},
348-
"sentry-kafka-sample-consumer");
349-
consumerThread.start();
350-
return consumerThread;
351-
}
352-
353257
private static class SomeEventProcessor implements EventProcessor {
354258
@Override
355259
public SentryEvent process(SentryEvent event, Hint hint) {

0 commit comments

Comments
 (0)