public class KubernetesJobAutoScalerContext implements JobAutoScalerContext<ResourceID> {

   private final ResourceID resourceID;

   private final JobID jobID;

   private final Configuration configuration;

   private final MetricGroup metricGroup;

   private final SupplierWithException<RestClusterClient<String>, Exception> restClientSupplier;

   private final AbstractFlinkResource<?, ?> resource;

   public KubernetesJobAutoScalerContext() {
           ResourceIDthis.resourceID = resourceID,;
       this.jobID   = JobID jobID,;
           Configurationthis.configuration = configuration,;
       this.metricGroup   = MetricGroup metricGroup,;
       this.restClientSupplier    SupplierWithException<RestClusterClient<String>, Exception> restClientSupplier,= restClientSupplier;
       this.resource = resource;
   AbstractFlinkResource<?, ?> resource) {}

   @Override
   public ResourceID getJobKey() {
 this.resourceID = resourceID;
    return resourceID;
  this.jobID = jobID; }

   @Override
   public this.configuration = configuration;JobID getJobID() {
       this.metricGroup = metricGroupreturn jobID;
   }

    this.restClientSupplier = restClientSupplier;@Override
   public Configuration getConfiguration() {
       this.resource = resourcereturn configuration;
   }

   @Override
   public ResourceIDMetricGroup getJobKeygetMetricGroup() {
       return resourceIDmetricGroup;
   }

   @Override
   public JobIDRestClusterClient<String> getJobIDgetRestClusterClient() throws Exception {
       return jobIDrestClientSupplier.get();
   }

   @Override
public AbstractFlinkResource<?,  public Configuration getConfiguration?> getResource() {
       return configurationresource;
   }

   @Override}

public class KubernetesAutoScalerEventHandler
     String reason,
 implements AutoScalerEventHandler<ResourceID, KubernetesJobAutoScalerContext> {

   private final EventRecorder eventRecorder;

   public KubernetesAutoScalerEventHandler(EventRecorder eventRecorder) {
       this.eventRecorder = eventRecorder;
  String }message,

   @Override
   public void handleEvent(KubernetesJobAutoScalerContext context,
                  @Nullable String messageKey) {
      Type type, eventRecorder.triggerEvent(
               context.getResource(),
            String reason   EventRecorder.Type.valueOf(type.name()),
               reason,
             String message  EventRecorder.Component.Operator,
               message,
            @Nullable  String messageKey) {;
   }

}

/** The kubernetes auto scaler state store, it's based on the config map. */ 
public class KubernetesAutoScalerStateStore
       implements AutoScalerStateStore<ResourceID, KubernetesJobAutoScalerContext> {

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

   private static final String LABEL_COMPONENT_AUTOSCALER = "autoscaler";
   private static final String COLLECTED_METRICS_KEY = "collectedMetrics";
   private static final String SCALING_HISTORY_KEY = "scalingHistory";
   private static final String PARALLELISM_OVERRIDES_KEY = "parallelismOverrides";

   private final KubernetesClient kubernetesClient;    this.kubernetesClient = kubernetesClient;
   }

   // COLLECTED_METRICS_KEY and PARALLELISM_OVERRIDES_KEY is similar to SCALING_HISTORY_KEY
   @Override
   public void storeScalingHistory(KubernetesJobAutoScalerContext jobContext, String decision) {
     getState(jobContext).put(SCALING_HISTORY_KEY, decision);
   }

   @Override
   public String getScalingHistory(KubernetesJobAutoScalerContext jobContext) {
    getState(jobContext).get(SCALING_HISTORY_KEY);
   }

   @Override
   public void removeScalingHistory(KubernetesJobAutoScalerContext jobContext) {
      getState(jobContext).remove(SCALING_HISTORY_KEY);
   }

   private final ConcurrentHashMap<ResourceIDMap<String, ConfigMap>String> cache = new ConcurrentHashMap<>();

getState(KubernetesJobAutoScalerContext jobContext) {
    public KubernetesAutoScalerStateStore(KubernetesClient kubernetesClient) {
       this.kubernetesClient = kubernetesClientreturn cache.computeIfAbsent(jobContext.getJobKey(), jobKey -> getOrCreateConfigMap(jobContext)).getData();
   }

   // COLLECTED_METRICS_KEY and PARALLELISM_OVERRIDES_KEY is similar to SCALING_HISTORY_KEY@Override
   public void flush(KubernetesJobAutoScalerContext jobContext) {
   @Override
   public void storeScalingHistory(KubernetesJobAutoScalerContext jobContext, String decision) {
ConfigMap configMap = cache.get(jobContext.getJobKey());
        getState(jobContext).put(SCALING_HISTORY_KEY, decision);
   }

   @OverridePreconditions.checkState(configMap != null, "The configMap shouldn't be null.");
       try {
   public String getScalingHistory(KubernetesJobAutoScalerContext jobContext) {
    getState cache.put(jobContext.getJobKey().get(SCALING_HISTORY_KEY, kubernetesClient.resource(configMap).update());
   }

   @Override
   public} voidcatch removeScalingHistory(KubernetesJobAutoScalerContextException jobContexte) {
      getState(jobContext).remove(SCALING_HISTORY_KEY);     LOG.error(
   }

   private Map<String, String> getState(KubernetesJobAutoScalerContext jobContext) {
       return cache.computeIfAbsent(jobContext.getJobKey(), jobKey -> getOrCreateConfigMap(jobContext)).getData();
   }

   @Override
   public void flush(KubernetesJobAutoScalerContext jobContext) {
 "Error while updating autoscaler info configmap, invalidating to clear the cache",
                ConfigMap configMap = cache.get(jobContext.getJobKey()) e);
       Preconditions.checkState(configMap != null, "The configMap shouldn't be null.");
    removeInfoFromCache(jobContext);
           throw try {e;
       }
    cache.put(jobContext.getJobKey(), kubernetesClient.resource(configMap).update());}

   @Override
   public }void catch removeInfoFromCache(ExceptionKubernetesJobAutoScalerContext ejobContext) {
           LOG.error(cache.remove(jobContext.getJobKey());
   }

   private ConfigMap getOrCreateConfigMap(KubernetesJobAutoScalerContext jobContext) {
       AbstractFlinkResource<?, ?> "Errorcr while updating autoscaler info configmap, invalidating to clear the cache",
  = jobContext.getResource();
       var meta = createCmObjectMeta(ResourceID.fromResource(cr));
                 ereturn getScalingInfoConfigMap(meta).orElseGet(() -> createConfigMap(cr, meta));
   }

   private ConfigMap createConfigMap(HasMetadata cr, ObjectMeta removeInfoFromCache(jobContext);meta) {
       LOG.info("Creating scaling info  throw econfig map");
       }
var cm  }

= new ConfigMap();
   @Override
   public void removeInfoFromCache(KubernetesJobAutoScalerContext jobContext) { cm.setMetadata(meta);
       cachecm.remove(jobContext.getJobKey())addOwnerReference(cr);
   }

   private ConfigMap getOrCreateConfigMap(KubernetesJobAutoScalerContext jobContext) {cm.setData(new HashMap<>());
       AbstractFlinkResource<?, ?> cr = jobContext.getResourcereturn kubernetesClient.resource(cm).create();
   }

   private varObjectMeta meta = createCmObjectMeta(ResourceID.fromResource(cr)); uid) {
       return getScalingInfoConfigMap(meta).orElseGet(() -> createConfigMap(cr, meta))var objectMeta = new ObjectMeta();
   }

   private ConfigMap createConfigMap(HasMetadata cr, ObjectMeta meta) { objectMeta.setName("autoscaler-" + uid.getName());
       LOG.info("Creating scaling info config map"uid.getNamespace().ifPresent(objectMeta::setNamespace);
       var cm = new ConfigMap();objectMeta.setLabels(
       cm.setMetadata(meta);
        cmMap.addOwnerReferenceof(cr);
       cm.setData(new HashMap<>());
       return kubernetesClient.resource(cm).create();
   }

   private ObjectMeta createCmObjectMeta(ResourceID uid) {
 Constants.LABEL_COMPONENT_KEY,
          var objectMeta = new ObjectMeta();
       objectMeta.setName("autoscaler-" + uid.getName());
 LABEL_COMPONENT_AUTOSCALER,
        uid.getNamespace().ifPresent(objectMeta::setNamespace);
       objectMeta.setLabels(
        Constants.LABEL_APP_KEY,
       Map.of(
                uid.getName()));
       Constants.LABEL_COMPONENT_KEY,return objectMeta;
   }

   private Optional<ConfigMap> getScalingInfoConfigMap(ObjectMeta objectMeta) {
       return Optional.ofNullable(
               LABEL_COMPONENT_AUTOSCALER,kubernetesClient
                       Constants.LABEL_APP_KEY,.configMaps()
                       uid.getNameinNamespace(objectMeta.getNamespace()));
       return objectMeta;
   }

   private Optional<ConfigMap> getScalingInfoConfigMap(ObjectMeta objectMeta) {
       return Optional.ofNullable(.withName(objectMeta.getName())
               kubernetesClient
                       .configMaps()
                       .inNamespace(objectMeta.getNamespace())
                       .withName(objectMeta.getName())
                       .get());
   }
}