Skip to content

Commit 16369e5

Browse files
wcheng92wcheng92
committed
[Update] Remove counting error codes for 200
1 parent fad6156 commit 16369e5

File tree

3 files changed

+5
-17
lines changed

3 files changed

+5
-17
lines changed

‎kafka-rest/src/main/java/io/confluent/kafkarest/response/StreamingResponse.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -74,7 +74,6 @@ public abstract class StreamingResponse<T> {
7474

7575
private static final Logger log = LoggerFactory.getLogger(StreamingResponse.class);
7676
private static final int ONE_SECOND_MS = 1000;
77-
private static final int HTTP_SUCCESS_CODE = 200;
7877

7978
private static final CompositeErrorMapper EXCEPTION_MAPPER =
8079
new CompositeErrorMapper.Builder()
@@ -228,10 +227,12 @@ private void closeAll(AsyncResponseQueue responseQueue) {
228227
private ResultOrError handleNext(
229228
T result, @Nullable Throwable error, ProduceRecordErrorCounter produceRecordErrorCounter) {
230229
if (error == null) {
231-
produceRecordErrorCounter.incrementErrorCount(HTTP_SUCCESS_CODE);
232230
return ResultOrError.result(result);
233231
} else {
234232
log.debug("Error processing streaming operation.", error);
233+
if (error.getCause() == null) {
234+
throw new IllegalArgumentException("Error cause is null", error);
235+
}
235236
int errorCode = EXCEPTION_MAPPER.toErrorResponse(error.getCause()).getErrorCode();
236237
produceRecordErrorCounter.incrementErrorCount(errorCode);
237238
return ResultOrError.error(EXCEPTION_MAPPER.toErrorResponse(error.getCause()));

‎kafka-rest/src/test/java/io/confluent/kafkarest/integration/CustomLogIntegrationTest.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -470,7 +470,7 @@ public void test_WhenProduceRequestWithMultipleRecords_ThenRequestLogHasAggregat
470470
false,
471471
true);
472472
} catch (Exception e) {
473-
e.printStackTrace();
473+
log.error("Failed to make produce request with multiple records", e);
474474
}
475475
}
476476

@@ -583,7 +583,7 @@ private void verifyLog(
583583
}
584584
if (isProduceRequestWithMultipleRecords) {
585585
rateLimitedLogEntry =
586-
"200 " + errorCodeInLog + " " + CustomLog.PRODUCE_ERROR_CODE_LOG_PREFIX + "200:5,429:5";
586+
"200 " + errorCodeInLog + " " + CustomLog.PRODUCE_ERROR_CODE_LOG_PREFIX + "429:5";
587587
}
588588
int rateLimitedRequests = 0;
589589
int totalRequests = 0;

‎kafka-rest/src/test/java/io/confluent/kafkarest/resources/v3/ProduceActionTest.java‎

Lines changed: 0 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -403,7 +403,6 @@ private void expectSetupStreamingRequestsWithExceptionsResults(
403403
expect(countLimitProvider.get()).andReturn(rateLimiterForCount);
404404
expect(bytesLimitProvider.get()).andReturn(rateLimiterForBytes);
405405
// first produce
406-
produceRecordErrorCounter.incrementErrorCount(200);
407406
expect(countLimiterGlobalProvider.get()).andReturn(countLimiterGlobal);
408407
expect(bytesLimiterGlobalProvider.get()).andReturn(bytesLimiterGlobal);
409408
bytesLimiterGlobal.rateLimit(anyInt());
@@ -422,7 +421,6 @@ private void expectSetupStreamingRequestsWithExceptionsResults(
422421
httpServletRequest.setAttribute(REST_ERROR_CODE, PRODUCE_MAX_BYTES_PER_TENANT_LIMIT_EXCEEDED);
423422
expectLastCall();
424423
// third produce
425-
produceRecordErrorCounter.incrementErrorCount(200);
426424
expect(countLimiterGlobalProvider.get()).andReturn(countLimiterGlobal);
427425
expect(bytesLimiterGlobalProvider.get()).andReturn(bytesLimiterGlobal);
428426
bytesLimiterGlobal.rateLimit(anyInt());
@@ -441,7 +439,6 @@ private void expectSetupStreamingRequestsWithExceptionsResults(
441439
REST_ERROR_CODE, PRODUCE_MAX_REQUESTS_PER_TENANT_LIMIT_EXCEEDED);
442440
expectLastCall();
443441
// fifth produce
444-
produceRecordErrorCounter.incrementErrorCount(200);
445442
expect(countLimiterGlobalProvider.get()).andReturn(countLimiterGlobal);
446443
expect(bytesLimiterGlobalProvider.get()).andReturn(bytesLimiterGlobal);
447444
bytesLimiterGlobal.rateLimit(anyInt());
@@ -460,7 +457,6 @@ private void expectSetupStreamingRequestsWithExceptionsResults(
460457
httpServletRequest.setAttribute(REST_ERROR_CODE, PRODUCE_MAX_BYTES_PER_TENANT_LIMIT_EXCEEDED);
461458
expectLastCall();
462459
// seventh produce
463-
produceRecordErrorCounter.incrementErrorCount(200);
464460
expect(countLimiterGlobalProvider.get()).andReturn(countLimiterGlobal);
465461
expect(bytesLimiterGlobalProvider.get()).andReturn(bytesLimiterGlobal);
466462
bytesLimiterGlobal.rateLimit(anyInt());
@@ -479,7 +475,6 @@ private void expectSetupStreamingRequestsWithExceptionsResults(
479475
httpServletRequest.setAttribute(REST_ERROR_CODE, PRODUCE_MAX_BYTES_PER_TENANT_LIMIT_EXCEEDED);
480476
expectLastCall();
481477
// ninth produce
482-
produceRecordErrorCounter.incrementErrorCount(200);
483478
expect(countLimiterGlobalProvider.get()).andReturn(countLimiterGlobal);
484479
expect(bytesLimiterGlobalProvider.get()).andReturn(bytesLimiterGlobal);
485480
bytesLimiterGlobal.rateLimit(anyInt());
@@ -533,7 +528,6 @@ public void streamingRequests() throws Exception {
533528

534529
expect(countLimitProvider.get()).andReturn(rateLimiterForCount);
535530
expect(bytesLimitProvider.get()).andReturn(rateLimiterForBytes);
536-
produceRecordErrorCounter.incrementErrorCount(200);
537531
expect(countLimiterGlobalProvider.get()).andReturn(countLimiterGlobal);
538532
expect(bytesLimiterGlobalProvider.get()).andReturn(bytesLimiterGlobal);
539533
rateLimiterForCount.rateLimit(anyInt());
@@ -542,21 +536,18 @@ public void streamingRequests() throws Exception {
542536
countLimiterGlobal.rateLimit(anyInt());
543537
rateLimiterForCount.rateLimit(anyInt());
544538
rateLimiterForBytes.rateLimit(anyInt());
545-
produceRecordErrorCounter.incrementErrorCount(200);
546539
expect(countLimiterGlobalProvider.get()).andReturn(countLimiterGlobal);
547540
expect(bytesLimiterGlobalProvider.get()).andReturn(bytesLimiterGlobal);
548541
bytesLimiterGlobal.rateLimit(anyInt());
549542
countLimiterGlobal.rateLimit(anyInt());
550543
rateLimiterForCount.rateLimit(anyInt());
551544
rateLimiterForBytes.rateLimit(anyInt());
552-
produceRecordErrorCounter.incrementErrorCount(200);
553545
expect(countLimiterGlobalProvider.get()).andReturn(countLimiterGlobal);
554546
expect(bytesLimiterGlobalProvider.get()).andReturn(bytesLimiterGlobal);
555547
bytesLimiterGlobal.rateLimit(anyInt());
556548
countLimiterGlobal.rateLimit(anyInt());
557549
rateLimiterForCount.rateLimit(anyInt());
558550
rateLimiterForBytes.rateLimit(anyInt());
559-
produceRecordErrorCounter.incrementErrorCount(200);
560551
expect(countLimiterGlobalProvider.get()).andReturn(countLimiterGlobal);
561552
expect(bytesLimiterGlobalProvider.get()).andReturn(bytesLimiterGlobal);
562553
bytesLimiterGlobal.rateLimit(anyInt());
@@ -662,7 +653,6 @@ public void produceWithByteLimit() throws Exception {
662653
bytesLimiterGlobal.rateLimit(anyInt());
663654
countLimiterGlobal.rateLimit(anyInt());
664655
ProduceRecordErrorCounter produceRecordErrorCounter1 = new ProduceRecordErrorCounter();
665-
produceRecordErrorCounter1.incrementErrorCount(200);
666656
httpServletRequest.setAttribute(
667657
EasyMock.eq(REST_PRODUCE_RECORD_ERROR_CODE_COUNTS),
668658
ProduceErrorCodeCounterMatcher.produceErrorCodeCounterIsMatched(
@@ -787,7 +777,6 @@ public void produceWithCountLimit() throws Exception {
787777
bytesLimiterGlobal.rateLimit(anyInt());
788778
countLimiterGlobal.rateLimit(anyInt());
789779
ProduceRecordErrorCounter produceRecordErrorCounter1 = new ProduceRecordErrorCounter();
790-
produceRecordErrorCounter1.incrementErrorCount(200);
791780
httpServletRequest.setAttribute(
792781
EasyMock.eq(REST_PRODUCE_RECORD_ERROR_CODE_COUNTS),
793782
ProduceErrorCodeCounterMatcher.produceErrorCodeCounterIsMatched(
@@ -933,7 +922,6 @@ public void produceNoLimit() throws Exception {
933922
mockedChunkedOutput.write(resultOrErrorOK1); // successful first produce
934923
mockedChunkedOutput.close();
935924
ProduceRecordErrorCounter produceRecordErrorCounter1 = new ProduceRecordErrorCounter();
936-
produceRecordErrorCounter1.incrementErrorCount(200);
937925
httpServletRequest.setAttribute(
938926
EasyMock.eq(REST_PRODUCE_RECORD_ERROR_CODE_COUNTS),
939927
ProduceErrorCodeCounterMatcher.produceErrorCodeCounterIsMatched(
@@ -946,7 +934,6 @@ public void produceNoLimit() throws Exception {
946934
mockedChunkedOutput.write(resultOrErrorOK2); // successful second produce
947935
mockedChunkedOutput.close();
948936
ProduceRecordErrorCounter produceRecordErrorCounter2 = new ProduceRecordErrorCounter();
949-
produceRecordErrorCounter2.incrementErrorCount(200);
950937
httpServletRequest.setAttribute(
951938
EasyMock.eq(REST_PRODUCE_RECORD_ERROR_CODE_COUNTS),
952939
ProduceErrorCodeCounterMatcher.produceErrorCodeCounterIsMatched(

0 commit comments

Comments
 (0)