Skip to content

AMBARI-26312: Clean up RequestScheduleEntity and RequestScheduleBatchRequestEntity also #3946

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: trunk
Choose a base branch
from
+97 −3
Open

AMBARI-26312: Clean up RequestScheduleEntity and RequestScheduleBatchRequestEntity also #3946

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
92 changes: 91 additions & 1 deletion ambari-server/src/main/java/org/apache/ambari/server/orm/dao/RequestScheduleDAO.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,25 +17,46 @@
*/
package org.apache.ambari.server.orm.dao;

import java.util.Date;
import java.util.List;

import javax.persistence.EntityManager;
import javax.persistence.NoResultException;
import javax.persistence.TypedQuery;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.cleanup.TimeBasedCleanupPolicy;
import org.apache.ambari.server.orm.RequiresSession;
import org.apache.ambari.server.orm.entities.RequestScheduleBatchRequestEntity;
import org.apache.ambari.server.orm.entities.RequestScheduleEntity;
import org.apache.ambari.server.state.Clusters;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.Singleton;
import com.google.inject.persist.Transactional;

@Singleton
public class RequestScheduleDAO {
public class RequestScheduleDAO implements Cleanable {

private static final Logger LOG = LoggerFactory.getLogger(RequestScheduleDAO.class);

@Inject
Provider<EntityManager> entityManagerProvider;

@Inject
private DaoUtils daoUtils;

@Inject
private Provider<Clusters> clusters;

/**
* Batch size to query the DB and use the results in an IN clause.
*/
private static final int BATCH_SIZE = 999;

@RequiresSession
public RequestScheduleEntity findById(Long id) {
return entityManagerProvider.get().find(RequestScheduleEntity.class, id);
Expand Down Expand Up @@ -89,4 +110,73 @@ public void removeByPK(Long id) {
public void refresh(RequestScheduleEntity requestScheduleEntity) {
entityManagerProvider.get().refresh(requestScheduleEntity);
}

/**
* Find all @RequestScheduleEntity with date before provided date.
* @param clusterId cluster id
* @param beforeDateMillis timestamp in millis
* @return List<Integer> ids
*/
private List<Integer> findAllScheduleIdsBeforeDate(Long clusterId, long beforeDateMillis) {

EntityManager entityManager = entityManagerProvider.get();
TypedQuery<Integer> requestScheduleQuery =
entityManager.createNamedQuery("RequestScheduleEntity.findAllReqScheduleIdsInClusterBeforeDate", Integer.class);

requestScheduleQuery.setParameter("clusterId", clusterId);
requestScheduleQuery.setParameter("beforeDate", beforeDateMillis);

return daoUtils.selectList(requestScheduleQuery);
}

/**
* Deletes RequestSchedule and RequestScheduleBatchRequest records in relation with RequestSchedule entries older than the given date.
*
* @param clusterId the identifier of the cluster the RequestSchedule belong to
* @param beforeDateMillis the date in milliseconds the
* @return a long representing the number of affected (deleted) records
*/
@Transactional
int cleanRequestSchedulesAndRequestScheduleBatchRequestsForClusterBeforeDate(Long clusterId, long beforeDateMillis) {
LOG.info("Deleting RequestSchedule and RequestScheduleBatchRequest entities before date " + new Date(beforeDateMillis));
EntityManager entityManager = entityManagerProvider.get();
List<Integer> ids = findAllScheduleIdsBeforeDate(clusterId, beforeDateMillis);
int affectedRows = 0;

TypedQuery<RequestScheduleEntity> requestScheduleQuery =
entityManager.createNamedQuery("RequestScheduleEntity.removeByScheduleIds", RequestScheduleEntity.class);
TypedQuery<RequestScheduleBatchRequestEntity> requestScheduleBatchRequestQuery =
entityManager.createNamedQuery("RequestScheduleBatchRequestEntity.removeByScheduleIds", RequestScheduleBatchRequestEntity.class);
if (ids != null && !ids.isEmpty()) {
for (int i = 0; i < ids.size(); i += BATCH_SIZE) {
int endIndex = Math.min((i + BATCH_SIZE), ids.size());
List<Integer> idsSubList = ids.subList(i, endIndex);
LOG.info("Deleting RequestScheduleBatchRequest entity batch with schedule ids: " +
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

RequestSchedule is related to RequestScheduleBatchRequest.
I guess that RequestSchedule is usually removed with RequestScheduleBatchRequest.

  @Override
  public void delete() {
    readWriteLock.writeLock().lock();
    try {
      if (isPersisted) {
        batchRequestDAO.removeByScheduleId(requestScheduleEntity.getScheduleId());
        requestScheduleDAO.remove(requestScheduleEntity);
        cluster.refresh();
        isPersisted = false;
      }
    } finally {
      readWriteLock.writeLock().unlock();
    }
  }

idsSubList.get(0) + " - " + idsSubList.get(idsSubList.size() - 1));
requestScheduleBatchRequestQuery.setParameter("scheduleIds", idsSubList);
affectedRows += requestScheduleBatchRequestQuery.executeUpdate();
LOG.info("Deleting RequestSchedule entity batch with schedule ids: " +
idsSubList.get(0) + " - " + idsSubList.get(idsSubList.size() - 1));
requestScheduleQuery.setParameter("scheduleIds", idsSubList);
affectedRows += requestScheduleQuery.executeUpdate();
}
}
return affectedRows;
}

@Transactional
@Override
public long cleanup(TimeBasedCleanupPolicy policy) {
long affectedRows = 0;
try {
Long clusterId = clusters.get().getCluster(policy.getClusterName()).getClusterId();
affectedRows += cleanRequestSchedulesAndRequestScheduleBatchRequestsForClusterBeforeDate(clusterId,
policy.getToDateInMillis());
} catch (AmbariException e) {
LOG.error("Error while looking up cluster with name: {}", policy.getClusterName(), e);
throw new IllegalStateException(e);
}

return affectedRows;
}
}
3 changes: 2 additions & 1 deletion ...rc/main/java/org/apache/ambari/server/orm/entities/RequestScheduleBatchRequestEntity.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,8 @@
@Table(name = "requestschedulebatchrequest")
@NamedQueries({
@NamedQuery(name = "findByScheduleId", query = "SELECT batchreqs FROM " +
"RequestScheduleBatchRequestEntity batchreqs WHERE batchreqs.scheduleId=:id")
"RequestScheduleBatchRequestEntity batchreqs WHERE batchreqs.scheduleId=:id"),
@NamedQuery(name = "RequestScheduleBatchRequestEntity.removeByScheduleIds", query = "DELETE FROM RequestScheduleBatchRequestEntity batchreqs WHERE batchreqs.scheduleId IN :scheduleIds")
})
public class RequestScheduleBatchRequestEntity {
@Id
Expand Down
5 changes: 4 additions & 1 deletion ambari-server/src/main/java/org/apache/ambari/server/orm/entities/RequestScheduleEntity.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,10 @@
"SELECT reqSchedule FROM RequestScheduleEntity reqSchedule"),
@NamedQuery(name = "reqScheduleByStatus", query =
"SELECT reqSchedule FROM RequestScheduleEntity reqSchedule " +
"WHERE reqSchedule.status=:status")
"WHERE reqSchedule.status=:status"),
@NamedQuery(name = "RequestScheduleEntity.findAllReqScheduleIdsInClusterBeforeDate", query =
"SELECT reqSchedule.scheduleId FROM RequestScheduleEntity reqSchedule WHERE reqSchedule.clusterId = :clusterId AND reqSchedule.updateTimestamp <= :beforeDate"),
@NamedQuery(name = "RequestScheduleEntity.removeByScheduleIds", query = "DELETE FROM RequestScheduleEntity reqSchedule WHERE reqSchedule.scheduleId IN :scheduleIds")
})
@TableGenerator(name = "schedule_id_generator",
table = "ambari_sequences", pkColumnName = "sequence_name", valueColumnName = "sequence_value"
Expand Down