Skip to content

Commit 29d010d

Browse files
libourel06libourel06
authored
Update TracingDecoratorTest.java
adapt unit test to new constructor of TracingDecorator
1 parent 1585505 commit 29d010d

File tree

1 file changed

+42
-39
lines changed

1 file changed

+42
-39
lines changed

‎impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/TracingDecoratorTest.java

+42-39
Original file line numberDiff line numberDiff line change
@@ -101,7 +101,7 @@
101101
public class TracingDecoratorTest {
102102
// header value format here: https://www.w3.org/TR/trace-context/#traceparent-header
103103
private static final String TRACE_PARENT = w3cHeader("1", "9");
104-
private static final String PROCESSOR_NAME = MockType.class.getName();
104+
private static final String PROCESSOR_NAME = io.quarkiverse.kafkastreamsprocessor.impl.decorator.processor.TracingDecoratorTest.MockType.class.getName();
105105
private static final Formatter LOG_FORMATTER = new PatternFormatter("%p %s %e");
106106
private static final InMemoryLogHandler inMemoryLogHandler = new InMemoryLogHandler(record -> true);
107107
private static final java.util.logging.Logger rootLogger = LogManager.getLogManager().getLogger("io.quarkiverse");
@@ -115,7 +115,7 @@ public class TracingDecoratorTest {
115115
JsonFormat.Printer jsonPrinter;
116116

117117
@Spy
118-
ReadMDCProcessor kafkaProcessor;
118+
io.quarkiverse.kafkastreamsprocessor.impl.decorator.processor.TracingDecoratorTest.ReadMDCProcessor kafkaProcessor;
119119

120120
@Mock
121121
InternalProcessorContext<String, Ping> processorContext;
@@ -134,9 +134,10 @@ public void setUp() {
134134
inMemoryLogHandler.getRecords().clear();
135135
rootLogger.addHandler(inMemoryLogHandler);
136136
rootLogger.setLevel(Level.DEBUG);
137-
when(topologyConfiguration.getProcessorPayloadType()).thenReturn((Class) MockType.class);
137+
when(topologyConfiguration.getProcessorPayloadType()).thenReturn((Class) io.quarkiverse.kafkastreamsprocessor.impl.decorator.processor.TracingDecoratorTest.MockType.class);
138138
decorator = new TracingDecorator(otel.getOpenTelemetry(), kafkaTextMapGetter,
139-
tracer, topologyConfiguration.getProcessorPayloadType().getName(), jsonPrinter);
139+
kafkaTextMapSetter, tracer, topologyConfiguration.getProcessorPayloadType().getName(),
140+
jsonPrinter);
140141
decorator.setDelegate(kafkaProcessor);
141142
decorator.init(processorContext);
142143
}
@@ -147,9 +148,9 @@ public void shouldSetMDCFromUberTraceId() {
147148
try (Scope parentScope = parentSpan.makeCurrent()) {
148149
Headers headers = new RecordHeaders();
149150
otel.getOpenTelemetry()
150-
.getPropagators()
151-
.getTextMapPropagator()
152-
.inject(Context.current(), headers, kafkaTextMapSetter);
151+
.getPropagators()
152+
.getTextMapPropagator()
153+
.inject(Context.current(), headers, kafkaTextMapSetter);
153154
Record<String, Ping> record = new Record<>(null, null, 0L, headers);
154155

155156
decorator.process(record);
@@ -166,13 +167,13 @@ public void shouldSetMDCFromUberTraceId() {
166167
public void shouldStartAndFinishSpan() {
167168
// manually build parent span to inject some TraceState and test the state is well recorded in the created span
168169
Span parentSpan = Span.wrap(SpanContext.create(IdGenerator.random().generateTraceId(), IdGenerator.random()
169-
.generateSpanId(), TraceFlags.getSampled(), TraceState.builder().put("state1", "value2").build()));
170+
.generateSpanId(), TraceFlags.getSampled(), TraceState.builder().put("state1", "value2").build()));
170171
try (Scope parentScope = parentSpan.makeCurrent()) {
171172
RecordHeaders headers = new RecordHeaders();
172173
otel.getOpenTelemetry()
173-
.getPropagators()
174-
.getTextMapPropagator()
175-
.inject(Context.current(), headers, kafkaTextMapSetter);
174+
.getPropagators()
175+
.getTextMapPropagator()
176+
.inject(Context.current(), headers, kafkaTextMapSetter);
176177
Record<String, Ping> record = new Record<>(null, null, 0L, headers);
177178

178179
decorator.process(record);
@@ -181,12 +182,12 @@ public void shouldStartAndFinishSpan() {
181182
}
182183

183184
assertThat(otel.getSpans())
184-
.hasTracesSatisfyingExactly(
185-
trace -> trace.hasSpansSatisfyingExactly(
186-
span -> span.hasTraceId(parentSpan.getSpanContext().getTraceId())
187-
.hasName(PROCESSOR_NAME)
188-
.hasParentSpanId(parentSpan.getSpanContext().getSpanId())
189-
.hasTraceState(TraceState.builder().put("state1", "value2").build())));
185+
.hasTracesSatisfyingExactly(
186+
trace -> trace.hasSpansSatisfyingExactly(
187+
span -> span.hasTraceId(parentSpan.getSpanContext().getTraceId())
188+
.hasName(PROCESSOR_NAME)
189+
.hasParentSpanId(parentSpan.getSpanContext().getSpanId())
190+
.hasTraceState(TraceState.builder().put("state1", "value2").build())));
190191
}
191192

192193
@Test
@@ -195,16 +196,17 @@ public void shouldCleanMDCAndScopeInCaseOfException() {
195196
try (Scope parentScope = parentSpan.makeCurrent()) {
196197
Headers headers = new RecordHeaders();
197198
otel.getOpenTelemetry()
198-
.getPropagators()
199-
.getTextMapPropagator()
200-
.inject(Context.current(), headers, kafkaTextMapSetter);
199+
.getPropagators()
200+
.getTextMapPropagator()
201+
.inject(Context.current(), headers, kafkaTextMapSetter);
201202
Record<String, Ping> record = new Record<>(null, Ping.newBuilder()
202-
.setMessage("blabla")
203-
.build(), 0L, headers);
203+
.setMessage("blabla")
204+
.build(), 0L, headers);
204205

205206
decorator = new TracingDecorator(otel.getOpenTelemetry(), kafkaTextMapGetter,
206-
tracer, topologyConfiguration.getProcessorPayloadType().getName(), jsonPrinter);
207-
decorator.setDelegate(new ThrowExceptionProcessor());
207+
kafkaTextMapSetter,
208+
tracer, topologyConfiguration.getProcessorPayloadType().getName(), jsonPrinter);
209+
decorator.setDelegate(new io.quarkiverse.kafkastreamsprocessor.impl.decorator.processor.TracingDecoratorTest.ThrowExceptionProcessor());
208210
decorator.init(processorContext);
209211

210212
assertDoesNotThrow(() -> decorator.process(record));
@@ -215,12 +217,12 @@ public void shouldCleanMDCAndScopeInCaseOfException() {
215217
assertNull(MDC.get("traceId"));
216218

217219
assertThat(otel.getSpans())
218-
.hasTracesSatisfyingExactly(trace -> trace.hasSpansSatisfyingExactly(
219-
span -> span.hasSpanId(parentSpan.getSpanContext().getSpanId()),
220-
span -> span.hasTraceId(parentSpan.getSpanContext().getTraceId())
221-
.hasName(PROCESSOR_NAME)
222-
.hasStatusSatisfying(status -> status.hasCode(StatusCode.ERROR))
223-
.hasException(new TestException())));
220+
.hasTracesSatisfyingExactly(trace -> trace.hasSpansSatisfyingExactly(
221+
span -> span.hasSpanId(parentSpan.getSpanContext().getSpanId()),
222+
span -> span.hasTraceId(parentSpan.getSpanContext().getTraceId())
223+
.hasName(PROCESSOR_NAME)
224+
.hasStatusSatisfying(status -> status.hasCode(StatusCode.ERROR))
225+
.hasException(new TestException())));
224226
}
225227

226228
@Test
@@ -257,7 +259,7 @@ void shouldManageRuntimeException() throws Throwable {
257259
decorator.process(new Record<>("key", inputMessage, 0L));
258260

259261
assertThat(getLogs(), hasItem(allOf(containsString("ERROR"),
260-
containsString("Runtime error caught while processing the message"), containsString(exception.getMessage()))));
262+
containsString("Runtime error caught while processing the message"), containsString(exception.getMessage()))));
261263
assertThat(getLogs(), hasItem(allOf(containsString("DEBUG"), containsString("marshalled"))));
262264
}
263265

@@ -269,7 +271,7 @@ private static List<String> getLogs() {
269271
void shouldLetBubbleUpKafkaExceptionAndLogMessage() {
270272
doThrow(new KafkaException()).when(kafkaProcessor).process(any());
271273
Assertions.assertThrows(KafkaException.class,
272-
() -> decorator.process(new Record<>("key", inputMessage, 0L)));
274+
() -> decorator.process(new Record<>("key", inputMessage, 0L)));
273275
}
274276

275277
@Test
@@ -296,7 +298,7 @@ void shouldLogMetadataEvenIfValueMarshallingToJSONFails() throws Throwable {
296298
decorator.process(new Record<>("key", inputMessage, 0L));
297299

298300
assertThat(getLogs(),
299-
hasItem(allOf(containsString("ERROR"), containsString(protocolBufferException.getMessage()))));
301+
hasItem(allOf(containsString("ERROR"), containsString(protocolBufferException.getMessage()))));
300302
assertThat(getLogs(), hasItem(allOf(containsString("DEBUG"), containsString("value=null"))));
301303
}
302304

@@ -305,7 +307,8 @@ void shouldLogRawToStringValueIfNotProtobuf() throws Throwable {
305307
Processor<String, String, String, String> kafkaProcessor = mock(Processor.class);
306308
ProcessorContext<String, String> processorContext = mock(ProcessorContext.class);
307309
TracingDecorator decorator = new TracingDecorator(GlobalOpenTelemetry.get(), kafkaTextMapGetter,
308-
tracer, topologyConfiguration.getProcessorPayloadType().getName(), jsonPrinter);
310+
kafkaTextMapSetter, tracer, topologyConfiguration.getProcessorPayloadType().getName(),
311+
jsonPrinter);
309312
decorator.setDelegate(kafkaProcessor);
310313
decorator.init(processorContext);
311314

@@ -322,11 +325,11 @@ void shouldLogRawToStringValueIfNotProtobuf() throws Throwable {
322325
void shouldPropagateOpentelemetryW3CBaggage() {
323326
// header value format here: https://www.w3.org/TR/baggage/#baggage-http-header-format
324327
Headers headers = new RecordHeaders().add(W3C_TRACE_ID, TRACE_PARENT.getBytes())
325-
.add(W3C_BAGGAGE, "key1=value1,key2=value2".getBytes());
328+
.add(W3C_BAGGAGE, "key1=value1,key2=value2".getBytes());
326329
Record<String, Ping> record = new Record<>(null, Ping.newBuilder().setMessage("blabla").build(), 0L, headers);
327-
decorator = new TracingDecorator(otel.getOpenTelemetry(), kafkaTextMapGetter,
328-
tracer, topologyConfiguration.getProcessorPayloadType().getName(), jsonPrinter);
329-
decorator.setDelegate(new LogOpentelemetryBaggageProcessor());
330+
decorator = new TracingDecorator(otel.getOpenTelemetry(), kafkaTextMapGetter, kafkaTextMapSetter,
331+
tracer, topologyConfiguration.getProcessorPayloadType().getName(), jsonPrinter);
332+
decorator.setDelegate(new io.quarkiverse.kafkastreamsprocessor.impl.decorator.processor.TracingDecoratorTest.LogOpentelemetryBaggageProcessor());
330333
decorator.init(processorContext);
331334

332335
decorator.process(record);
@@ -363,7 +366,7 @@ public void process(Record<String, Ping> record) {
363366

364367
public static String w3cHeader(String traceId, String spanId) {
365368
return String.format("00-%s-%s-01", StringUtils.leftPad(traceId, TraceId.getLength(), '0'),
366-
StringUtils.leftPad(spanId, SpanId.getLength(), '0'));
369+
StringUtils.leftPad(spanId, SpanId.getLength(), '0'));
367370
}
368371

369372
public static class MockType {

0 commit comments

Comments
 (0)