Removed check_fail_status in FastenKafkaConsumer
@@ -190,70 +190,6 @@ public class FastenKafkaConsumer extends FastenKafkaConnection {
return statusRecords;
}
- /**
- * It checks whether the last record was processed successfully or not.
- * if the processing of the last record failed, the method changes current offset to skip the failed record.
- */
- private void checkFailStatus() {
-
- ArrayList<TopicPartition> topicPartitions = new ArrayList<>();
- for (PartitionInfo p : this.cgsStatusConsumer.partitionsFor(this.failedRecordsTopic)) {
- topicPartitions.add(new TopicPartition(this.failedRecordsTopic, p.partition()));
- }
-
- ConsumerRecords<String, String> statusRecords = dummyPoll(this.cgsStatusConsumer);
-// ConsumerRecords<String, String> statusRecords;
-// int i = 0;
-// do{
-// statusRecords = this.cgsStatusConsumer.poll(Duration.ofMillis(100));
-// i++;
-// } while (i <= 5 && statusRecords.count() == 0);
- // After five times try, either we continue to check or assume that CGS_status topic is empty and return.
- if (statusRecords.count() == 0) return;
-
- this.cgsStatusConsumer.seekToEnd(topicPartitions);
-
- //System.out.println("Offset before seeking to end: " + this.cgsStatusConsumer.position(topicPartitions.get(0)));
-
- // Sets offset to correct position to read the last record
- topicPartitions.forEach(tp -> this.cgsStatusConsumer.seek(tp, this.cgsStatusConsumer.position(tp) - 1));
- //this.cgsStatusConsumer.seek(topicPartitions.get(0), this.cgsStatusConsumer.position(topicPartitions.get(0)) - 1);
-
- //System.out.println("Offset set to: " + this.cgsStatusConsumer.position(topicPartitions.get(0)));
-
- statusRecords = this.cgsStatusConsumer.poll(Duration.ofMillis(100));
-
- if (statusRecords.count() != 0) {
- String record = statusRecords.iterator().next().value();
- String[] splitRecord = record.split("\\|");
- JSONObject lastStatusRecord = new JSONObject(splitRecord[1]);
-
- //System.out.println("Checking " + splitRecord[1]);
- logger.debug("Checking the status of record: " + splitRecord[1]);
-
- this.cgsStatusConsumer.close();
-
- if (!lastStatusRecord.get("status").toString().equals(this.OK_STATUS)) {
- logger.debug("Increasing offset for skipping a record");
- final long failedPartition = Long.parseLong(lastStatusRecord.get("partition").toString());
- final long failedOffset = Long.parseLong(lastStatusRecord.get("offset").toString());
-
- ConsumerRecords<String, String> failedRecord;
- do {
- // A dummy call to poll to make seek method work
- failedRecord = this.connection.poll(Duration.ofMillis(1000));
- connection.commitSync();
- System.out.println("Fetched " + failedRecord.count());
- } while (failedRecord.count() == 0);
-
- System.out.println("Current offset for plug-in" + this.connection.position(new TopicPartition(lastStatusRecord.get("topic").toString(),
- (int) failedPartition)));
- this.connection.seek(new TopicPartition(lastStatusRecord.get("topic").toString(), (int) failedPartition),
- failedOffset + 1);
- }
- }
- }
-
/**
* This method adds one to the offset of all the partitions of a topic.
* This is useful when you want to skip an offset with FATAL errors when the FASTEN server is restarted
