kimyongin/RefreshByManualTask.java

## RefreshByManualTask.java
package subscription.refresh.manager.task;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomStringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import slack.sender.service.SlackService;
import stamp.common.utils.DateTimeUtil;
import stamp.dao.dynamo.m.constants.ComparisonOperator;
import stamp.dao.dynamo.m.constants.TableColumns;
import stamp.dao.dynamo.m.constants.TableNames;
import stamp.dao.dynamo.m.entity.SubscriptionEntity;
import stamp.dao.dynamo.m.repository.SubscriptionRepository;
import stamp.dao.dynamo.m.service.ShardDaemonService;
import stamp.dao.dynamo.m.util.QueryScanExpressionBuilder;
import subscription.refresh.manager.service.SubscriptionRefreshService;

import javax.annotation.PostConstruct;
import java.text.MessageFormat;
import java.util.List;

import static stamp.dao.dynamo.m.util.QueryScanExpressionBuilder.OrderBy.Ascending;

@Slf4j
@Component
@RequiredArgsConstructor
public class RefreshByManualTask {
  @Value("${shard-service-id.subscription-refresh}")
  private String serviceId;
  private final String daemonId = RandomStringUtils.randomAlphanumeric(10);
  private final ShardDaemonService shardDaemonService;
  private final SubscriptionRepository subscriptionRepository;
  private final SubscriptionRefreshService subscriptionRefreshService;
  private final SlackService slackService;

  /**
  schedule.health-check-ms=60000
  schedule.adjust-ms=60000
  schedule.consume-ms=60000
  retire-shard-delete-ms=60000
  shard-query-max-size=100
  schedule.pulling-message=1000
  manage-shard-count=1
  **/

  @Value("${schedule.health-check-ms}")
  private long healthCheckIntervalMs;
  @Value("${schedule.consume-ms}")
  private long consumeIntervalMs;
  @Value("${retire-shard-delete-ms}")
  private long retireShardDeleteMs;
  @Value("${shard-query-max-size}")
  private int shardQueryMaxSize;
  @Value("${manage-shard-count}")
  private int manageShardCount;

  @PostConstruct
  public void postConstruct() {
    shardDaemonService.initialize(serviceId, daemonId, this::hasShardData);
  }

  @Scheduled(fixedRateString = "${schedule.health-check-ms}")
  protected void onHealthCheck() {
    shardDaemonService.healthCheck();
  }

  @Scheduled(fixedDelayString = "${schedule.adjust-ms}")
  protected void onAdjust() {
    int myShardCount = shardDaemonService.adjust(healthCheckIntervalMs * 3, consumeIntervalMs * 3, retireShardDeleteMs);
    if (myShardCount > manageShardCount) {
      slackService.sendMessage(MessageFormat.format("{0} - {1} has {2} shard.", this.serviceId, this.daemonId, myShardCount));
    }
  }

  @Scheduled(fixedDelayString = "${schedule.consume-ms}")
  protected void onConsume() {
    List<String> shardIds = shardDaemonService.getOwnedShardIds();
    shardIds.forEach(shardDaemonService::updateLastConsumed);

    shardIds.forEach(shardId -> {
      List<SubscriptionEntity> subscriptionEntities = querySubscriptionToRefresh(shardId, shardQueryMaxSize);
      subscriptionEntities.forEach(subscriptionRefreshService::handleScheduledCheck);
    });
  }

  private boolean hasShardData(String shardId) {
    QueryScanExpressionBuilder queryExpression = QueryScanExpressionBuilder.builder()
        .indexName(TableNames.SUBSCRIPTION_GSI_SCHEDULE_GROUP_AND_SCHEDULE_DATE_MS)
        .hashKey(TableColumns.SCHEDULE_GROUP, shardId)
        .limit(1);
    return subscriptionRepository.query(queryExpression).size() > 0;
  }

  private List<SubscriptionEntity> querySubscriptionToRefresh(String shardId, int size) {
    QueryScanExpressionBuilder queryExpression = QueryScanExpressionBuilder.builder()
        .indexName(TableNames.SUBSCRIPTION_GSI_SCHEDULE_GROUP_AND_SCHEDULE_DATE_MS)
        .hashKey(TableColumns.SCHEDULE_GROUP, shardId)
        .rangeKeyCompare(TableColumns.SCHEDULE_DATE_MS, ComparisonOperator.LT, DateTimeUtil.nowUtc0LongMs())
        .limit(size)
        .rangeOrderBy(Ascending);
    return subscriptionRepository.query(queryExpression);
  }

  private void handleFail(Exception ex, SubscriptionEntity subscription) {
    log.info("[REFRESH_TASK] id - {}, key - {}, handle fail - {}", subscription.getSubscriptionId(), subscription.getSubscriptionKey(), ex.toString());
  }
}

## sharddaemon.java
package stamp.dao.dynamo.m.service;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import stamp.common.utils.DateTimeUtil;
import stamp.dao.dynamo.m.constants.TableColumns;
import stamp.dao.dynamo.m.entity.DaemonEntity;
import stamp.dao.dynamo.m.entity.ShardEntity;
import stamp.dao.dynamo.m.repository.DaemonRepository;
import stamp.dao.dynamo.m.repository.ShardRepository;
import stamp.dao.dynamo.m.util.QueryScanExpressionBuilder;
import stamp.dao.dynamo.m.util.SaveExpressionBuilder;

import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.function.Function;
import java.util.stream.Collectors;

@Slf4j
@Service
@RequiredArgsConstructor
public class ShardDaemonService {
  private final ShardRepository stampShardRepository;
  private final DaemonRepository stampDaemonRepository;

  private DaemonEntity me;
  private List<ShardEntity> myShards = new ArrayList<>();
  private Function<String, Boolean> hasShardData;

  public void initialize(String serviceId, String daemonId, Function<String, Boolean> hasShardData) {
    SaveExpressionBuilder saveExpressionBuilder = SaveExpressionBuilder.builder().notExists(TableColumns.DAEMON_ID);
    me = DaemonEntity.builder()
        .serviceId(serviceId)
        .daemonId(daemonId)
        .lastHealthCheckMs((DateTimeUtil.nowUtc0LongMs()))
        .build();
    stampDaemonRepository.save(me, saveExpressionBuilder);
    this.hasShardData = hasShardData;
  }

  public void healthCheck() {
    me.setLastHealthCheckMs(DateTimeUtil.nowUtc0LongMs());
    stampDaemonRepository.save(me);
    log.info(MessageFormat.format("[{0}][{1}] healthCheck", me.getServiceId(), me.getDaemonId()));
  }

  public int adjust(long healthCheckMaxDelayMs, long consumeMaxDelayMs, long retireShardDeleteMs) {
    long now = DateTimeUtil.nowUtc0LongMs();
    QueryScanExpressionBuilder queryByServiceId = QueryScanExpressionBuilder.builder().hashKey(TableColumns.SERVICE_ID, me.getServiceId());
    List<DaemonEntity> daemons = stampDaemonRepository.query(queryByServiceId);
    List<ShardEntity> shards = stampShardRepository.query(queryByServiceId);

    // Daemon 분류 : Healthy, Unhealthy
    List<DaemonEntity> healthyDaemons = new ArrayList<>();
    List<DaemonEntity> unhealthyDaemons = new ArrayList<>();
    daemons.forEach(daemon -> {
      if (now - daemon.getLastHealthCheckMs() < healthCheckMaxDelayMs) {
        healthyDaemons.add(daemon);
      } else {
        unhealthyDaemons.add(daemon);
      }
    });

    // Delete Unhealthy Daemon
    unhealthyDaemons.forEach(daemon -> deleteDaemon(daemon, "unhealthy"));

    myShards = new ArrayList<>();
    if (shards.size() == 0) {
      return -1; // 처리할 샤드가 없다.
    }

    // Shard 분류 : MyShard, OtherShard, FreeShard
    List<ShardEntity> freeShards = new ArrayList<>();
    List<ShardEntity> otherShards = new ArrayList<>();
    List<ShardEntity> retireShards = new ArrayList<>();
    shards.forEach(shard -> {
      String daemonId = shard.getOwnerDaemonId();
      if (daemonId == null) {
        freeShards.add(shard);
      } else if (Objects.equals(me.getDaemonId(), daemonId)) {
        myShards.add(shard);
      } else if (!Objects.equals(me.getDaemonId(), daemonId)) {
        otherShards.add(shard);
      }
      if (shard.getRetiredDateMs() != null) {
        retireShards.add(shard);
      }
    });

    // Release Delay Shard
    otherShards.forEach(otherShard -> {
      if (now - otherShard.getLastConsumedDateMs() > consumeMaxDelayMs) {
        DaemonEntity daemon = stampDaemonRepository.load(otherShard.getServiceId(), otherShard.getOwnerDaemonId());
        if (daemon == null || now - daemon.getLastHealthCheckMs() > healthCheckMaxDelayMs) {
          releaseShard(otherShard, "delayed");
        }
      }
    });

    // Delete Retire Shard
    retireShards.forEach(retireShard -> {
      if (now - retireShard.getRetiredDateMs() > retireShardDeleteMs && !hasShardData.apply(retireShard.getShardId())) {
        deleteShard(retireShard, "retired");
      }
    });

    // Re balance(Acquire or Release) Shard
    float myShardCount = myShards.size();
    float idealShardCount = (float) shards.size() / (float) healthyDaemons.size();
    float balance = myShardCount - idealShardCount;
    if (balance >= 1) {
      List<ShardEntity> releaseShards = myShards.stream()
          .peek(shard -> releaseShard(shard, "exceed balance"))
          .limit((long) balance)
          .collect(Collectors.toList());
      myShards.removeAll(releaseShards);
      log.info(MessageFormat.format("[{0}][{1}] releaseCount : {2}", me.getServiceId(), me.getDaemonId(), releaseShards.size()));
    } else if (balance < 0 && freeShards.size() > 0) {
      List<ShardEntity> acquireShards = freeShards.stream()
          .peek(shard -> acquireShard(shard, "under balance"))
          .limit((long) Math.ceil(Math.abs(balance)))
          .collect(Collectors.toList());
      myShards.addAll(acquireShards);
      log.info(MessageFormat.format("[{0}][{1}] acquireCount : {2}", me.getServiceId(), me.getDaemonId(), acquireShards.size()));
    }

    return CollectionUtils.isEmpty(myShards) ? 0 : myShards.size();
  }

  public List<String> getOwnedShardIds() {
    return myShards.stream()
        .map(ShardEntity::getShardId)
        .collect(Collectors.toList());
  }

  public void updateLastConsumed(String shardId) {
    Optional<ShardEntity> shardEntityOpt = myShards.stream().filter(entity -> Objects.equals(entity.getShardId(), shardId)).findFirst();
    if (shardEntityOpt.isPresent()) {
      ShardEntity stampShardEntity = shardEntityOpt.get();
      stampShardEntity.setLastConsumedDateMs(DateTimeUtil.nowUtc0LongMs());
      stampShardRepository.save(stampShardEntity);
    }
  }

  private void acquireShard(ShardEntity stampShardEntity, String reason) {
    SaveExpressionBuilder saveCondition = SaveExpressionBuilder.builder().notExists(TableColumns.OWNER_DAEMON_ID);
    stampShardEntity.setOwnerDaemonId(me.getDaemonId());
    stampShardEntity.setLastConsumedDateMs(DateTimeUtil.nowUtc0LongMs());
    stampShardRepository.save(stampShardEntity, saveCondition);
    log.info(MessageFormat.format("[{0}][{1}] acquireShard {2} because {3}", me.getServiceId(), me.getDaemonId(), stampShardEntity.getShardId(), reason));
  }

  private void releaseShard(ShardEntity stampShardEntity, String reason) {
    SaveExpressionBuilder saveCondition = SaveExpressionBuilder.builder().eq(TableColumns.OWNER_DAEMON_ID, stampShardEntity.getOwnerDaemonId());
    stampShardEntity.setOwnerDaemonId(null);
    stampShardRepository.save(stampShardEntity, saveCondition);
    log.info(MessageFormat.format("[{0}][{1}] releaseShard {2} because {3}", me.getServiceId(), me.getDaemonId(), stampShardEntity.getShardId(), reason));
  }

  private void deleteDaemon(DaemonEntity stampDaemonEntity, String reason) {
    stampDaemonRepository.delete(stampDaemonEntity);
    log.info(MessageFormat.format("[{0}][{1}] deleteDaemon because {2}", me.getServiceId(), me.getDaemonId(), reason));
  }

  private void deleteShard(ShardEntity stampShardEntity, String reason) {
    stampShardRepository.delete(stampShardEntity);
    log.info(MessageFormat.format("[{0}][{1}] deleteShard {2} because {3}", me.getServiceId(), me.getDaemonId(), stampShardEntity.getShardId(), reason));
  }
}
	package subscription.refresh.manager.task;

	import lombok.RequiredArgsConstructor;
	import lombok.extern.slf4j.Slf4j;
	import org.apache.commons.lang3.RandomStringUtils;
	import org.springframework.beans.factory.annotation.Value;
	import org.springframework.scheduling.annotation.Scheduled;
	import org.springframework.stereotype.Component;
	import slack.sender.service.SlackService;
	import stamp.common.utils.DateTimeUtil;
	import stamp.dao.dynamo.m.constants.ComparisonOperator;
	import stamp.dao.dynamo.m.constants.TableColumns;
	import stamp.dao.dynamo.m.constants.TableNames;
	import stamp.dao.dynamo.m.entity.SubscriptionEntity;
	import stamp.dao.dynamo.m.repository.SubscriptionRepository;
	import stamp.dao.dynamo.m.service.ShardDaemonService;
	import stamp.dao.dynamo.m.util.QueryScanExpressionBuilder;
	import subscription.refresh.manager.service.SubscriptionRefreshService;

	import javax.annotation.PostConstruct;
	import java.text.MessageFormat;
	import java.util.List;

	import static stamp.dao.dynamo.m.util.QueryScanExpressionBuilder.OrderBy.Ascending;

	@Slf4j
	@Component
	@RequiredArgsConstructor
	public class RefreshByManualTask {
	@Value("${shard-service-id.subscription-refresh}")
	private String serviceId;
	private final String daemonId = RandomStringUtils.randomAlphanumeric(10);
	private final ShardDaemonService shardDaemonService;
	private final SubscriptionRepository subscriptionRepository;
	private final SubscriptionRefreshService subscriptionRefreshService;
	private final SlackService slackService;

	/**
	schedule.health-check-ms=60000
	schedule.adjust-ms=60000
	schedule.consume-ms=60000
	retire-shard-delete-ms=60000
	shard-query-max-size=100
	schedule.pulling-message=1000
	manage-shard-count=1
	**/

	@Value("${schedule.health-check-ms}")
	private long healthCheckIntervalMs;
	@Value("${schedule.consume-ms}")
	private long consumeIntervalMs;
	@Value("${retire-shard-delete-ms}")
	private long retireShardDeleteMs;
	@Value("${shard-query-max-size}")
	private int shardQueryMaxSize;
	@Value("${manage-shard-count}")
	private int manageShardCount;

	@PostConstruct
	public void postConstruct() {
	shardDaemonService.initialize(serviceId, daemonId, this::hasShardData);
	}

	@Scheduled(fixedRateString = "${schedule.health-check-ms}")
	protected void onHealthCheck() {
	shardDaemonService.healthCheck();
	}

	@Scheduled(fixedDelayString = "${schedule.adjust-ms}")
	protected void onAdjust() {
	int myShardCount = shardDaemonService.adjust(healthCheckIntervalMs * 3, consumeIntervalMs * 3, retireShardDeleteMs);
	if (myShardCount > manageShardCount) {
	slackService.sendMessage(MessageFormat.format("{0} - {1} has {2} shard.", this.serviceId, this.daemonId, myShardCount));
	}
	}

	@Scheduled(fixedDelayString = "${schedule.consume-ms}")
	protected void onConsume() {
	List<String> shardIds = shardDaemonService.getOwnedShardIds();
	shardIds.forEach(shardDaemonService::updateLastConsumed);

	shardIds.forEach(shardId -> {
	List<SubscriptionEntity> subscriptionEntities = querySubscriptionToRefresh(shardId, shardQueryMaxSize);
	subscriptionEntities.forEach(subscriptionRefreshService::handleScheduledCheck);
	});
	}

	private boolean hasShardData(String shardId) {
	QueryScanExpressionBuilder queryExpression = QueryScanExpressionBuilder.builder()
	.indexName(TableNames.SUBSCRIPTION_GSI_SCHEDULE_GROUP_AND_SCHEDULE_DATE_MS)
	.hashKey(TableColumns.SCHEDULE_GROUP, shardId)
	.limit(1);
	return subscriptionRepository.query(queryExpression).size() > 0;
	}

	private List<SubscriptionEntity> querySubscriptionToRefresh(String shardId, int size) {
	QueryScanExpressionBuilder queryExpression = QueryScanExpressionBuilder.builder()
	.indexName(TableNames.SUBSCRIPTION_GSI_SCHEDULE_GROUP_AND_SCHEDULE_DATE_MS)
	.hashKey(TableColumns.SCHEDULE_GROUP, shardId)
	.rangeKeyCompare(TableColumns.SCHEDULE_DATE_MS, ComparisonOperator.LT, DateTimeUtil.nowUtc0LongMs())
	.limit(size)
	.rangeOrderBy(Ascending);
	return subscriptionRepository.query(queryExpression);
	}

	private void handleFail(Exception ex, SubscriptionEntity subscription) {
	log.info("[REFRESH_TASK] id - {}, key - {}, handle fail - {}", subscription.getSubscriptionId(), subscription.getSubscriptionKey(), ex.toString());
	}
	}
	package stamp.dao.dynamo.m.service;

	import lombok.RequiredArgsConstructor;
	import lombok.extern.slf4j.Slf4j;
	import org.springframework.stereotype.Service;
	import org.springframework.util.CollectionUtils;
	import stamp.common.utils.DateTimeUtil;
	import stamp.dao.dynamo.m.constants.TableColumns;
	import stamp.dao.dynamo.m.entity.DaemonEntity;
	import stamp.dao.dynamo.m.entity.ShardEntity;
	import stamp.dao.dynamo.m.repository.DaemonRepository;
	import stamp.dao.dynamo.m.repository.ShardRepository;
	import stamp.dao.dynamo.m.util.QueryScanExpressionBuilder;
	import stamp.dao.dynamo.m.util.SaveExpressionBuilder;

	import java.text.MessageFormat;
	import java.util.ArrayList;
	import java.util.List;
	import java.util.Objects;
	import java.util.Optional;
	import java.util.function.Function;
	import java.util.stream.Collectors;

	@Slf4j
	@Service
	@RequiredArgsConstructor
	public class ShardDaemonService {
	private final ShardRepository stampShardRepository;
	private final DaemonRepository stampDaemonRepository;

	private DaemonEntity me;
	private List<ShardEntity> myShards = new ArrayList<>();
	private Function<String, Boolean> hasShardData;

	public void initialize(String serviceId, String daemonId, Function<String, Boolean> hasShardData) {
	SaveExpressionBuilder saveExpressionBuilder = SaveExpressionBuilder.builder().notExists(TableColumns.DAEMON_ID);
	me = DaemonEntity.builder()
	.serviceId(serviceId)
	.daemonId(daemonId)
	.lastHealthCheckMs((DateTimeUtil.nowUtc0LongMs()))
	.build();
	stampDaemonRepository.save(me, saveExpressionBuilder);
	this.hasShardData = hasShardData;
	}

	public void healthCheck() {
	me.setLastHealthCheckMs(DateTimeUtil.nowUtc0LongMs());
	stampDaemonRepository.save(me);
	log.info(MessageFormat.format("[{0}][{1}] healthCheck", me.getServiceId(), me.getDaemonId()));
	}

	public int adjust(long healthCheckMaxDelayMs, long consumeMaxDelayMs, long retireShardDeleteMs) {
	long now = DateTimeUtil.nowUtc0LongMs();
	QueryScanExpressionBuilder queryByServiceId = QueryScanExpressionBuilder.builder().hashKey(TableColumns.SERVICE_ID, me.getServiceId());
	List<DaemonEntity> daemons = stampDaemonRepository.query(queryByServiceId);
	List<ShardEntity> shards = stampShardRepository.query(queryByServiceId);

	// Daemon 분류 : Healthy, Unhealthy
	List<DaemonEntity> healthyDaemons = new ArrayList<>();
	List<DaemonEntity> unhealthyDaemons = new ArrayList<>();
	daemons.forEach(daemon -> {
	if (now - daemon.getLastHealthCheckMs() < healthCheckMaxDelayMs) {
	healthyDaemons.add(daemon);
	} else {
	unhealthyDaemons.add(daemon);
	}
	});

	// Delete Unhealthy Daemon
	unhealthyDaemons.forEach(daemon -> deleteDaemon(daemon, "unhealthy"));

	myShards = new ArrayList<>();
	if (shards.size() == 0) {
	return -1; // 처리할 샤드가 없다.
	}

	// Shard 분류 : MyShard, OtherShard, FreeShard
	List<ShardEntity> freeShards = new ArrayList<>();
	List<ShardEntity> otherShards = new ArrayList<>();
	List<ShardEntity> retireShards = new ArrayList<>();
	shards.forEach(shard -> {
	String daemonId = shard.getOwnerDaemonId();
	if (daemonId == null) {
	freeShards.add(shard);
	} else if (Objects.equals(me.getDaemonId(), daemonId)) {
	myShards.add(shard);
	} else if (!Objects.equals(me.getDaemonId(), daemonId)) {
	otherShards.add(shard);
	}
	if (shard.getRetiredDateMs() != null) {
	retireShards.add(shard);
	}
	});

	// Release Delay Shard
	otherShards.forEach(otherShard -> {
	if (now - otherShard.getLastConsumedDateMs() > consumeMaxDelayMs) {
	DaemonEntity daemon = stampDaemonRepository.load(otherShard.getServiceId(), otherShard.getOwnerDaemonId());
	if (daemon == null \|\| now - daemon.getLastHealthCheckMs() > healthCheckMaxDelayMs) {
	releaseShard(otherShard, "delayed");
	}
	}
	});

	// Delete Retire Shard
	retireShards.forEach(retireShard -> {
	if (now - retireShard.getRetiredDateMs() > retireShardDeleteMs && !hasShardData.apply(retireShard.getShardId())) {
	deleteShard(retireShard, "retired");
	}
	});

	// Re balance(Acquire or Release) Shard
	float myShardCount = myShards.size();
	float idealShardCount = (float) shards.size() / (float) healthyDaemons.size();
	float balance = myShardCount - idealShardCount;
	if (balance >= 1) {
	List<ShardEntity> releaseShards = myShards.stream()
	.peek(shard -> releaseShard(shard, "exceed balance"))
	.limit((long) balance)
	.collect(Collectors.toList());
	myShards.removeAll(releaseShards);
	log.info(MessageFormat.format("[{0}][{1}] releaseCount : {2}", me.getServiceId(), me.getDaemonId(), releaseShards.size()));
	} else if (balance < 0 && freeShards.size() > 0) {
	List<ShardEntity> acquireShards = freeShards.stream()
	.peek(shard -> acquireShard(shard, "under balance"))
	.limit((long) Math.ceil(Math.abs(balance)))
	.collect(Collectors.toList());
	myShards.addAll(acquireShards);
	log.info(MessageFormat.format("[{0}][{1}] acquireCount : {2}", me.getServiceId(), me.getDaemonId(), acquireShards.size()));
	}

	return CollectionUtils.isEmpty(myShards) ? 0 : myShards.size();
	}

	public List<String> getOwnedShardIds() {
	return myShards.stream()
	.map(ShardEntity::getShardId)
	.collect(Collectors.toList());
	}

	public void updateLastConsumed(String shardId) {
	Optional<ShardEntity> shardEntityOpt = myShards.stream().filter(entity -> Objects.equals(entity.getShardId(), shardId)).findFirst();
	if (shardEntityOpt.isPresent()) {
	ShardEntity stampShardEntity = shardEntityOpt.get();
	stampShardEntity.setLastConsumedDateMs(DateTimeUtil.nowUtc0LongMs());
	stampShardRepository.save(stampShardEntity);
	}
	}

	private void acquireShard(ShardEntity stampShardEntity, String reason) {
	SaveExpressionBuilder saveCondition = SaveExpressionBuilder.builder().notExists(TableColumns.OWNER_DAEMON_ID);
	stampShardEntity.setOwnerDaemonId(me.getDaemonId());
	stampShardEntity.setLastConsumedDateMs(DateTimeUtil.nowUtc0LongMs());
	stampShardRepository.save(stampShardEntity, saveCondition);
	log.info(MessageFormat.format("[{0}][{1}] acquireShard {2} because {3}", me.getServiceId(), me.getDaemonId(), stampShardEntity.getShardId(), reason));
	}

	private void releaseShard(ShardEntity stampShardEntity, String reason) {
	SaveExpressionBuilder saveCondition = SaveExpressionBuilder.builder().eq(TableColumns.OWNER_DAEMON_ID, stampShardEntity.getOwnerDaemonId());
	stampShardEntity.setOwnerDaemonId(null);
	stampShardRepository.save(stampShardEntity, saveCondition);
	log.info(MessageFormat.format("[{0}][{1}] releaseShard {2} because {3}", me.getServiceId(), me.getDaemonId(), stampShardEntity.getShardId(), reason));
	}

	private void deleteDaemon(DaemonEntity stampDaemonEntity, String reason) {
	stampDaemonRepository.delete(stampDaemonEntity);
	log.info(MessageFormat.format("[{0}][{1}] deleteDaemon because {2}", me.getServiceId(), me.getDaemonId(), reason));
	}

	private void deleteShard(ShardEntity stampShardEntity, String reason) {
	stampShardRepository.delete(stampShardEntity);
	log.info(MessageFormat.format("[{0}][{1}] deleteShard {2} because {3}", me.getServiceId(), me.getDaemonId(), stampShardEntity.getShardId(), reason));
	}
	}