Skip to content

Commit 571dba4

Browse files
otelbot[bot]otelbot[bot]
committed
./gradlew spotlessApply
1 parent ee88ff0 commit 571dba4

File tree

7 files changed

+32
-26
lines changed

7 files changed

+32
-26
lines changed

‎instrumentation/kafka/kafka-connect-2.6/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaconnect/v2_6/KafkaConnectAttributesGetter.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -91,7 +91,7 @@ public List<String> getMessageHeader(KafkaConnectTask request, String name) {
9191
if (firstRecord == null || firstRecord.headers() == null) {
9292
return Collections.emptyList();
9393
}
94-
94+
9595
return StreamSupport.stream(firstRecord.headers().spliterator(), false)
9696
.filter(header -> name.equals(header.key()) && header.value() != null)
9797
.map(header -> convertHeaderValue(header))

‎instrumentation/kafka/kafka-connect-2.6/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaconnect/v2_6/KafkaConnectBatchProcessSpanLinksExtractor.java‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -13,11 +13,12 @@
1313
import org.apache.kafka.connect.sink.SinkRecord;
1414

1515
/**
16-
* Extracts span links from Kafka Connect SinkRecord headers for batch processing scenarios.
17-
* This ensures that when processing a batch of records that may come from different traces,
18-
* we create links to all the original trace contexts rather than losing them.
16+
* Extracts span links from Kafka Connect SinkRecord headers for batch processing scenarios. This
17+
* ensures that when processing a batch of records that may come from different traces, we create
18+
* links to all the original trace contexts rather than losing them.
1919
*/
20-
final class KafkaConnectBatchProcessSpanLinksExtractor implements SpanLinksExtractor<KafkaConnectTask> {
20+
final class KafkaConnectBatchProcessSpanLinksExtractor
21+
implements SpanLinksExtractor<KafkaConnectTask> {
2122

2223
private final SpanLinksExtractor<SinkRecord> singleRecordLinkExtractor;
2324

@@ -27,8 +28,7 @@ final class KafkaConnectBatchProcessSpanLinksExtractor implements SpanLinksExtra
2728
}
2829

2930
@Override
30-
public void extract(
31-
SpanLinksBuilder spanLinks, Context parentContext, KafkaConnectTask request) {
31+
public void extract(SpanLinksBuilder spanLinks, Context parentContext, KafkaConnectTask request) {
3232

3333
for (SinkRecord record : request.getRecords()) {
3434
// Create a link to each record's original trace context

‎instrumentation/kafka/kafka-connect-2.6/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaconnect/v2_6/KafkaConnectSingletons.java‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,15 +27,18 @@ public final class KafkaConnectSingletons {
2727
private static final Instrumenter<KafkaConnectTask, Void> INSTRUMENTER;
2828

2929
static {
30-
KafkaConnectBatchProcessSpanLinksExtractor spanLinksExtractor = new KafkaConnectBatchProcessSpanLinksExtractor(PROPAGATOR);
30+
KafkaConnectBatchProcessSpanLinksExtractor spanLinksExtractor =
31+
new KafkaConnectBatchProcessSpanLinksExtractor(PROPAGATOR);
3132

3233
INSTRUMENTER =
3334
Instrumenter.<KafkaConnectTask, Void>builder(
3435
GlobalOpenTelemetry.get(),
3536
INSTRUMENTATION_NAME,
36-
MessagingSpanNameExtractor.create(KafkaConnectAttributesGetter.INSTANCE, MessageOperation.PROCESS))
37+
MessagingSpanNameExtractor.create(
38+
KafkaConnectAttributesGetter.INSTANCE, MessageOperation.PROCESS))
3739
.addAttributesExtractor(
38-
MessagingAttributesExtractor.builder(KafkaConnectAttributesGetter.INSTANCE, MessageOperation.PROCESS)
40+
MessagingAttributesExtractor.builder(
41+
KafkaConnectAttributesGetter.INSTANCE, MessageOperation.PROCESS)
3942
.build())
4043
.addSpanLinksExtractor(spanLinksExtractor)
4144
.buildInstrumenter(SpanKindExtractor.alwaysConsumer());

‎instrumentation/kafka/kafka-connect-2.6/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaconnect/v2_6/KafkaConnectTask.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -21,9 +21,9 @@ public Collection<SinkRecord> getRecords() {
2121
}
2222

2323
/**
24-
* Returns the first record in the batch, used for extracting destination information.
25-
* Kafka Connect processes records in batches, but all records in a batch typically
26-
* come from the same topic, so we use the first record for span naming.
24+
* Returns the first record in the batch, used for extracting destination information. Kafka
25+
* Connect processes records in batches, but all records in a batch typically come from the same
26+
* topic, so we use the first record for span naming.
2727
*/
2828
public SinkRecord getFirstRecord() {
2929
return records.isEmpty() ? null : records.iterator().next();

‎instrumentation/kafka/kafka-connect-2.6/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaconnect/v2_6/SinkRecordHeadersGetter.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -13,8 +13,8 @@
1313
import org.apache.kafka.connect.sink.SinkRecord;
1414

1515
/**
16-
* Extracts trace context from Kafka Connect SinkRecord headers for distributed tracing.
17-
* This enables proper trace propagation from the original producer through Kafka Connect processing.
16+
* Extracts trace context from Kafka Connect SinkRecord headers for distributed tracing. This
17+
* enables proper trace propagation from the original producer through Kafka Connect processing.
1818
*/
1919
enum SinkRecordHeadersGetter implements TextMapGetter<SinkRecord> {
2020
INSTANCE;
@@ -24,7 +24,7 @@ public Iterable<String> keys(SinkRecord record) {
2424
if (record.headers() == null) {
2525
return java.util.Collections.emptyList();
2626
}
27-
27+
2828
return StreamSupport.stream(record.headers().spliterator(), false)
2929
.map(Header::key)
3030
.collect(java.util.stream.Collectors.toList());

‎instrumentation/kafka/kafka-connect-2.6/testing/src/test/java/io/opentelemetry/instrumentation/kafkaconnect/v2_6/MongoKafkaConnectSinkTaskTest.java‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -127,7 +127,7 @@ private static String getKafkaBoostrapServers() {
127127

128128
@BeforeAll
129129
public static void setup() throws IOException {
130-
130+
131131
// Create log directory on desktop
132132
File logDir = new File(System.getProperty("user.home") + "/Desktop/kafka-connect-logs");
133133
if (!logDir.exists()) {
@@ -227,8 +227,8 @@ public static void setup() throws IOException {
227227
new Slf4jLogConsumer(LoggerFactory.getLogger("kafka-connect-container")))
228228
// Save logs to desktop
229229
.withFileSystemBind(
230-
System.getProperty("user.home") + "/Desktop/kafka-connect-logs",
231-
"/var/log/kafka-connect",
230+
System.getProperty("user.home") + "/Desktop/kafka-connect-logs",
231+
"/var/log/kafka-connect",
232232
BindMode.READ_WRITE)
233233
// Copy the agent jar to the container
234234
.withCopyFileToContainer(
@@ -270,11 +270,11 @@ public static void setup() throws IOException {
270270
.withCommand(
271271
"bash",
272272
"-c",
273-
"mkdir -p /var/log/kafka-connect && " +
274-
"confluent-hub install --no-prompt --component-dir /usr/share/java " +
275-
"mongodb/kafka-connect-mongodb:1.11.0 && " +
276-
"echo 'Starting Kafka Connect with logging to /var/log/kafka-connect/' && " +
277-
"/etc/confluent/docker/run 2>&1 | tee /var/log/kafka-connect/kafka-connect.log");
273+
"mkdir -p /var/log/kafka-connect && "
274+
+ "confluent-hub install --no-prompt --component-dir /usr/share/java "
275+
+ "mongodb/kafka-connect-mongodb:1.11.0 && "
276+
+ "echo 'Starting Kafka Connect with logging to /var/log/kafka-connect/' && "
277+
+ "/etc/confluent/docker/run 2>&1 | tee /var/log/kafka-connect/kafka-connect.log");
278278

279279
mongoDB =
280280
new MongoDBContainer(DockerImageName.parse("mongo:4.4"))

‎instrumentation/kafka/kafka-connect-2.6/testing/src/test/java/io/opentelemetry/instrumentation/kafkaconnect/v2_6/PostgresKafkaConnectSinkTaskTest.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -148,7 +148,7 @@ private static int getRandomFreePort() {
148148

149149
@BeforeAll
150150
public static void setup() throws IOException {
151-
151+
152152
// Create log directory on desktop
153153
File logDir = new File(System.getProperty("user.home") + "/Desktop/kafka-connect-logs");
154154
if (!logDir.exists()) {
@@ -407,7 +407,10 @@ public void testKafkaConnectSinkTaskInstrumentation() throws IOException, Interr
407407
foundKafkaConnectSpan = true;
408408
if (spanIdNode != null) {
409409
kafkaConnectSpanId = spanIdNode.asText();
410-
logger.info("Found Kafka Connect span '{}' with ID: {}", spanName, kafkaConnectSpanId);
410+
logger.info(
411+
"Found Kafka Connect span '{}' with ID: {}",
412+
spanName,
413+
kafkaConnectSpanId);
411414
}
412415
}
413416

0 commit comments

Comments
 (0)