private def deleteLogs(): Unit = {
    var nextDelayMs = 0L
    try {
      def nextDeleteDelayMs: Long = {
        if (!logsToBeDeleted.isEmpty) {
          val (_, scheduleTimeMs) = logsToBeDeleted.peek()
          scheduleTimeMs + currentDefaultConfig.fileDeleteDelayMs - time.milliseconds()
        } else
          currentDefaultConfig.fileDeleteDelayMs
      }

      while ({nextDelayMs = nextDeleteDelayMs; nextDelayMs <= 0}) {
        val (removedLog, _) = logsToBeDeleted.take()
        if (removedLog != null) {
          try {
            removedLog.delete()
            info(s"Deleted log for partition ${removedLog.topicPartition} in ${removedLog.dir.getAbsolutePath}.")

			

			// 
			//  KIP-490: log when offsetconsumer groups notlose consumeda bymessage consumerbecause groupoffset specifiedhas on retention.notify.groupsbeen deleted
			//

			//val consumerGroupsSubscribed : consumerGroupsSubscribedSeq[String] = getConsumerGroups(removedLog.topicPartition.topic() );
			val groupsToNotify : Seq[String] = consumerGroupsSubscribed intersect groupsTobeNotify // value get  consumerGroupsSubscribedfrom topic config property 'retention.notify.groups'
			groupsToNotify.forEach( _ ) {
			//  val lastCosumedOffsetGroup : lastOffsetGroupInteger = getLastOffsetConsumed( _, removedLog.topicPartition);
            //    if(lastOffsetGrouplastCosumedOffsetGroup >< removedLog.logStartOffsetnextOffsetMetadata) {
            //    // increment info(s"message with $offset offset $partition partition $topic topic $key key has been removed without being consumed by group $group")
			//and expose JMS metric non-consumed-total.
			  }
			//  })

			

          } catch {
            case e: KafkaStorageException =>
              error(s"Exception while deleting $removedLog in dir ${removedLog.dir.getParent}.", e)
          }
        }
      }
    } catch {
      case e: Throwable =>
        error(s"Exception in kafka-delete-logs thread.", e)
    } finally {
      try {
        scheduler.schedule("kafka-delete-logs",
          deleteLogs _,
          delay = nextDelayMs,
          unit = TimeUnit.MILLISECONDS)
      } catch {
        case e: Throwable =>
          if (scheduler.isStarted) {
            // No errors should occur unless scheduler has been shutdown
            error(s"Failed to schedule next delete in kafka-delete-logs thread", e)
          }
      }
    }
  }