pricelees/understanding-kafka-consumer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: '연속' 판단을 위해 비어있지 않은 레코드를 입력받는 경우 count를 0으로 설정하도록 기능 수정

Browse Source
This commit is contained in:
이상진 2025-06-27 16:08:45 +09:00
parent ebd003b79e
commit e5916eb305
1 changed files with 4 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/java/com/sangdol/consumer/infrastructure/kafka/consumer/RecordProcessor.java
View File

@ -22,7 +22,7 @@ public class RecordProcessor {
private long initialRecordReceivedTime = 0; private long initialRecordReceivedTime = 0;
public void process(ConsumerRecords<String, TestRecord> records) throws Exception { public void process(ConsumerRecords<String, TestRecord> records) throws Exception {
initializeTime(); preProcess();
log.debug("[RecordProcessor] process {} records. thread: {}", records.count(), log.debug("[RecordProcessor] process {} records. thread: {}", records.count(),
Thread.currentThread().getName()); Thread.currentThread().getName());
@ -50,9 +50,11 @@ public class RecordProcessor {
} }
} }
private void initializeTime() { private void preProcess() {
if (initialRecordReceivedTime == 0) { if (initialRecordReceivedTime == 0) {
initialRecordReceivedTime = System.currentTimeMillis(); initialRecordReceivedTime = System.currentTimeMillis();
} }
emptyPollResultCount.set(0);
} }
} }