src/main/java/com/sangdol/consumer/infrastructure/kafka/consumer/ConsumerWorker.java

@@ -10,6 +10,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.consumer.OffsetCommitCallback;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.WakeupException;
 
@@ -47,6 +48,38 @@ public class ConsumerWorker implements Runnable {
 		} catch (Exception e) {
 			log.info("[ConsumerWorker] Unexpected Exception. thread: {}", Thread.currentThread().getName());
 		} finally {
+			closeAfterManualCommit();
+		}
+	}
+
+	public void shutdown() {
+		running.set(false);
+		consumer.wakeup();
+	}
+
+	private void commitAsync(ConsumerRecords<String, TestRecord> records) {
+		Map<TopicPartition, OffsetAndMetadata> offsets = records.partitions().stream()
+			.collect(Collectors.toMap(tp -> tp, tp -> {
+				List<ConsumerRecord<String, TestRecord>> partitionRecords = records.records(tp);
+				long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
+
+				return new OffsetAndMetadata(lastOffset + 1);
+			}));
+
+		OffsetCommitCallback callback = (o, exception) -> {
+			if (exception != null) {
+				log.error("[ConsumerWorker] failed to commit offsets for offset: {}, partition {}", o,
+					records.partitions(), exception);
+			} else {
+				log.debug("[ConsumerWorker] committed offsets for offset: {}, partition {}", o,
+					records.partitions());
+			}
+		};
+
+		consumer.commitAsync(offsets, callback);
+	}
+
+	private void closeAfterManualCommit() {
 		try {
 			consumer.commitSync();
 			log.info("[ConsumerWorker] Final offset committed. Close consumer.. ");
@@ -59,29 +92,3 @@ public class ConsumerWorker implements Runnable {
 		}
 	}
 }
-
-	private void commitAsync(ConsumerRecords<String, TestRecord> records) {
-		Map<TopicPartition, OffsetAndMetadata> offsets = records.partitions().stream()
-			.collect(Collectors.toMap(tp -> tp, tp -> {
-				List<ConsumerRecord<String, TestRecord>> partitionRecords = records.records(tp);
-				long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
-
-				return new OffsetAndMetadata(lastOffset + 1);
-			}));
-
-		consumer.commitAsync(offsets, (o, exception) -> {
-			if (exception != null) {
-				log.error("[ConsumerWorker] failed to commit offsets for offset: {}, partition {}", o,
-					records.partitions(), exception);
-			} else {
-				log.debug("[ConsumerWorker] committed offsets for offset: {}, partition {}", o,
-					records.partitions());
-			}
-		});
-	}
-
-	public void shutdown() {
-		running.set(false);
-		consumer.wakeup();
-	}
-}