Skip to content

Commit 5455c46

Browse files
loicgreffierloicgreffier
authored
Fix Sonar issues in store path matching and tests (#545)
1 parent 0803eee commit 5455c46

5 files changed

Lines changed: 47 additions & 48 deletions

File tree

‎kstreamplify-core/src/main/java/com/michelin/kstreamplify/server/KafkaStreamsHttpServer.java‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -259,15 +259,15 @@ private Object getResponseForStoreEndpoints(HttpExchange exchange) {
259259
// Get all for timestamped key-value store
260260
if (exchange.getRequestURI()
261261
.toString()
262-
.matches("/" + DEFAULT_STORE_PATH + "/" + DEFAULT_KEY_VALUE_STORE_PATH + "/timestamped/.*")) {
262+
.matches("/" + DEFAULT_STORE_PATH + "/" + DEFAULT_KEY_VALUE_STORE_PATH + "/timestamped/[^/]+")) {
263263
store = parsePathParam(exchange, 4);
264264
return timestampedKeyValueService.getAll(store);
265265
}
266266

267267
// Get by key for key-value store
268268
if (exchange.getRequestURI()
269269
.toString()
270-
.matches("/" + DEFAULT_STORE_PATH + "/" + DEFAULT_KEY_VALUE_STORE_PATH + "/.*/.*")) {
270+
.matches("/" + DEFAULT_STORE_PATH + "/" + DEFAULT_KEY_VALUE_STORE_PATH + "/[^/]+/[^/]+")) {
271271
store = parsePathParam(exchange, 3);
272272
String key = parsePathParam(exchange, 4);
273273
return keyValueService.getByKey(store, key);
@@ -276,15 +276,15 @@ private Object getResponseForStoreEndpoints(HttpExchange exchange) {
276276
// Get all for key-value store
277277
if (exchange.getRequestURI()
278278
.toString()
279-
.matches("/" + DEFAULT_STORE_PATH + "/" + DEFAULT_KEY_VALUE_STORE_PATH + "/.*")) {
279+
.matches("/" + DEFAULT_STORE_PATH + "/" + DEFAULT_KEY_VALUE_STORE_PATH + "/[^/]+")) {
280280
store = parsePathParam(exchange, 3);
281281
return keyValueService.getAll(store);
282282
}
283283

284284
// Get by key for timestamped window store
285285
if (exchange.getRequestURI()
286286
.toString()
287-
.matches("/" + DEFAULT_STORE_PATH + "/" + DEFAULT_WINDOW_STORE_PATH + "/timestamped/.*/.*")) {
287+
.matches("/" + DEFAULT_STORE_PATH + "/" + DEFAULT_WINDOW_STORE_PATH + "/timestamped/[^/]+/[^/]+")) {
288288
store = parsePathParam(exchange, 4);
289289
String key = parsePathParam(exchange, 5);
290290
Instant instantFrom = parseRequestParam(exchange, START_TIME_REQUEST_PARAM)
@@ -300,7 +300,7 @@ private Object getResponseForStoreEndpoints(HttpExchange exchange) {
300300
// Get all for timestamped window store
301301
if (exchange.getRequestURI()
302302
.toString()
303-
.matches("/" + DEFAULT_STORE_PATH + "/" + DEFAULT_WINDOW_STORE_PATH + "/timestamped/.*")) {
303+
.matches("/" + DEFAULT_STORE_PATH + "/" + DEFAULT_WINDOW_STORE_PATH + "/timestamped/[^/]+")) {
304304
store = parsePathParam(exchange, 4);
305305
Instant instantFrom = parseRequestParam(exchange, START_TIME_REQUEST_PARAM)
306306
.map(Instant::parse)
@@ -315,7 +315,7 @@ private Object getResponseForStoreEndpoints(HttpExchange exchange) {
315315
// Get by key for window store
316316
if (exchange.getRequestURI()
317317
.toString()
318-
.matches("/" + DEFAULT_STORE_PATH + "/" + DEFAULT_WINDOW_STORE_PATH + "/.*/.*")) {
318+
.matches("/" + DEFAULT_STORE_PATH + "/" + DEFAULT_WINDOW_STORE_PATH + "/[^/]+/[^/]+")) {
319319
store = parsePathParam(exchange, 3);
320320
String key = parsePathParam(exchange, 4);
321321
Instant instantFrom = parseRequestParam(exchange, START_TIME_REQUEST_PARAM)
@@ -331,7 +331,7 @@ private Object getResponseForStoreEndpoints(HttpExchange exchange) {
331331
// Get all for window store
332332
if (exchange.getRequestURI()
333333
.toString()
334-
.matches("/" + DEFAULT_STORE_PATH + "/" + DEFAULT_WINDOW_STORE_PATH + "/.*")) {
334+
.matches("/" + DEFAULT_STORE_PATH + "/" + DEFAULT_WINDOW_STORE_PATH + "/[^/]+")) {
335335
store = parsePathParam(exchange, 3);
336336
Instant instantFrom = parseRequestParam(exchange, START_TIME_REQUEST_PARAM)
337337
.map(Instant::parse)

‎kstreamplify-core/src/test/java/com/michelin/kstreamplify/integration/container/KafkaIntegrationTest.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -153,8 +153,8 @@ private static String schemaRegistryUrl() {
153153

154154
public static <K, V> void produceRecordToTopic(List<ProducerRecord<K, V>> records, Properties properties) {
155155
try (KafkaProducer<K, V> producer = new KafkaProducer<>(properties)) {
156-
for (ProducerRecord<K, V> record : records) {
157-
producer.send(record);
156+
for (ProducerRecord<K, V> producerRecord : records) {
157+
producer.send(producerRecord);
158158
}
159159
producer.flush();
160160
}
@@ -170,8 +170,8 @@ public static <K, V> List<ConsumerRecord<K, V>> readAllRecordsFromTopic(
170170
while (totalPollTimeMs < 30000 && consumerRecords.size() < expectedNumberOfRecords) {
171171
totalPollTimeMs += pollIntervalMs;
172172
final ConsumerRecords<K, V> records = consumer.poll(Duration.ofMillis(pollIntervalMs));
173-
for (final ConsumerRecord<K, V> record : records) {
174-
consumerRecords.add(record);
173+
for (final ConsumerRecord<K, V> consumerRecord : records) {
174+
consumerRecords.add(consumerRecord);
175175
}
176176
}
177177
return consumerRecords;

‎kstreamplify-core/src/test/java/com/michelin/kstreamplify/integration/error/DlqProductionExceptionHandlerIntegrationTest.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -145,7 +145,6 @@ public void topology(StreamsBuilder streamsBuilder) {
145145
.to(
146146
"OUTPUT_TOPIC",
147147
Produced.with(Serdes.String(), Serdes.String()).withName("sink-processor"));
148-
;
149148
}
150149

151150
@Override

‎kstreamplify-core/src/test/java/com/michelin/kstreamplify/serde/TopicWithSerdeTest.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -286,8 +286,8 @@ public void init(ProcessorContext<Void, Void> context) {
286286
}
287287

288288
@Override
289-
public void process(Record<String, String> record) {
290-
store.put(record.key(), record.value());
289+
public void process(Record<String, String> message) {
290+
store.put(message.key(), message.value());
291291
}
292292
}
293293

@@ -300,8 +300,8 @@ public void init(ProcessorContext<Void, Void> context) {
300300
}
301301

302302
@Override
303-
public void process(Record<String, KafkaUserStub> record) {
304-
store.put(record.key(), ValueAndTimestamp.make(record.value(), record.timestamp()));
303+
public void process(Record<String, KafkaUserStub> message) {
304+
store.put(message.key(), ValueAndTimestamp.make(message.value(), message.timestamp()));
305305
}
306306
}
307307
}

‎kstreamplify-spring-boot/src/test/java/com/michelin/kstreamplify/integration/error/DlqProcessingExceptionHandlerIntegrationTest.java‎

Lines changed: 32 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -74,11 +74,12 @@ void setUp() throws InterruptedException {
7474
@Test
7575
void shouldSendRecordToDlqWhenProcessingFails() {
7676
ProducerRecord<String, String> message = new ProducerRecord<>("STRING_TOPIC", "user", "Doe");
77-
ProducerRecord<String, String> error_1 =
77+
ProducerRecord<String, String> errorMessage1 =
7878
new ProducerRecord<>("STRING_TOPIC", "key-error-1", "Filter Exception");
79-
ProducerRecord<String, String> error_2 = new ProducerRecord<>("STRING_TOPIC", "Map Exception", "value-error-2");
79+
ProducerRecord<String, String> errorMessage2 =
80+
new ProducerRecord<>("STRING_TOPIC", "Map Exception", "value-error-2");
8081
List<ProducerRecord<String, String>> records =
81-
List.of(message, message, error_1, message, message, error_2, message);
82+
List.of(message, message, errorMessage1, message, message, errorMessage2, message);
8283
produceRecordToTopic(records, getKafkaGlobalProperties());
8384

8485
Properties properties = getKafkaGlobalProperties();
@@ -92,44 +93,43 @@ void shouldSendRecordToDlqWhenProcessingFails() {
9293
List<ConsumerRecord<String, String>> consumerRecords = readAllRecordsFromTopic("OUTPUT_TOPIC", properties, 5);
9394
assertEquals(5, consumerRecords.size());
9495

95-
KafkaError kafkaError_1 = dlqConsumerRecords.get(0).value();
96+
KafkaError kafkaError1 = dlqConsumerRecords.get(0).value();
9697
assertEquals(
9798
"An exception occurred during the stream processing of a record. Please find more details about the exception in the cause and stack fields.",
98-
kafkaError_1.getContextMessage());
99-
assertEquals(2, kafkaError_1.getOffset());
100-
assertEquals(2, kafkaError_1.getPartition());
101-
assertEquals("STRING_TOPIC", kafkaError_1.getTopic());
102-
assertEquals("appKeyValueProcessingExceptionHandlerId", kafkaError_1.getApplicationId());
103-
assertEquals("filter-values", kafkaError_1.getProcessorNodeId());
104-
assertEquals("0_2", kafkaError_1.getTaskId());
105-
assertEquals("Exception while filtering values...", kafkaError_1.getCause());
106-
assertTrue(kafkaError_1.getStack().contains("java.lang.RuntimeException: Wrapper"));
107-
assertEquals("Filter Exception", new String(kafkaError_1.getByteValue().array()));
108-
assertEquals("key-error-1", new String(kafkaError_1.getSourceRawKey().array()));
99+
kafkaError1.getContextMessage());
100+
assertEquals(2, kafkaError1.getOffset());
101+
assertEquals(2, kafkaError1.getPartition());
102+
assertEquals("STRING_TOPIC", kafkaError1.getTopic());
103+
assertEquals("appKeyValueProcessingExceptionHandlerId", kafkaError1.getApplicationId());
104+
assertEquals("filter-values", kafkaError1.getProcessorNodeId());
105+
assertEquals("0_2", kafkaError1.getTaskId());
106+
assertEquals("Exception while filtering values...", kafkaError1.getCause());
107+
assertTrue(kafkaError1.getStack().contains("java.lang.RuntimeException: Wrapper"));
108+
assertEquals("Filter Exception", new String(kafkaError1.getByteValue().array()));
109+
assertEquals("key-error-1", new String(kafkaError1.getSourceRawKey().array()));
109110
assertEquals(
110-
"Filter Exception", new String(kafkaError_1.getSourceRawValue().array()));
111-
assertNull(kafkaError_1.getValue());
111+
"Filter Exception", new String(kafkaError1.getSourceRawValue().array()));
112+
assertNull(kafkaError1.getValue());
112113

113-
KafkaError kafkaError_2 = dlqConsumerRecords.get(1).value();
114+
KafkaError kafkaError2 = dlqConsumerRecords.get(1).value();
114115
assertEquals(
115116
"An exception occurred during the stream processing of a record. Please find more details about the exception in the cause and stack fields.",
116-
kafkaError_1.getContextMessage());
117-
assertEquals(0, kafkaError_2.getOffset());
118-
assertEquals(0, kafkaError_2.getPartition());
117+
kafkaError1.getContextMessage());
118+
assertEquals(0, kafkaError2.getOffset());
119+
assertEquals(0, kafkaError2.getPartition());
119120
assertEquals(
120-
"appKeyValueProcessingExceptionHandlerId-repartitioned-values-repartition", kafkaError_2.getTopic());
121-
assertEquals("appKeyValueProcessingExceptionHandlerId", kafkaError_2.getApplicationId());
122-
assertEquals("map-values", kafkaError_2.getProcessorNodeId());
123-
assertEquals("1_0", kafkaError_2.getTaskId());
124-
assertEquals("Exception while mapping values...", kafkaError_2.getCause());
125-
assertTrue(kafkaError_1.getStack().contains("java.lang.RuntimeException: Wrapper"));
121+
"appKeyValueProcessingExceptionHandlerId-repartitioned-values-repartition", kafkaError2.getTopic());
122+
assertEquals("appKeyValueProcessingExceptionHandlerId", kafkaError2.getApplicationId());
123+
assertEquals("map-values", kafkaError2.getProcessorNodeId());
124+
assertEquals("1_0", kafkaError2.getTaskId());
125+
assertEquals("Exception while mapping values...", kafkaError2.getCause());
126+
assertTrue(kafkaError1.getStack().contains("java.lang.RuntimeException: Wrapper"));
126127
assertEquals(
127128
"transformed-value-error-2",
128-
new String(kafkaError_2.getByteValue().array()));
129-
assertEquals("Map Exception", new String(kafkaError_2.getSourceRawKey().array()));
130-
assertEquals(
131-
"value-error-2", new String(kafkaError_2.getSourceRawValue().array()));
132-
assertNull(kafkaError_2.getValue());
129+
new String(kafkaError2.getByteValue().array()));
130+
assertEquals("Map Exception", new String(kafkaError2.getSourceRawKey().array()));
131+
assertEquals("value-error-2", new String(kafkaError2.getSourceRawValue().array()));
132+
assertNull(kafkaError2.getValue());
133133
}
134134

135135
/**

0 commit comments

Comments
 (0)