package com.s24.redjob.worker.json;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.SerializationException;

import com.fasterxml.jackson.databind.Module;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.s24.redjob.worker.Execution;
import com.s24.redjob.worker.NoResult;

/**
 * {@link RedisSerializer} for {@link Execution}s.
 */
public class ExecutionRedisSerializer extends Jackson2JsonRedisSerializer<Execution> {
   /**
    * Logger.
    */
   private static final Logger log = LoggerFactory.getLogger(ExecutionRedisSerializer.class);

   /**
    * Object mapper.
    */
   private ObjectMapper objectMapper = new ObjectMapper();

   /**
    * Ignore deserialization failures?.
    */
   private boolean ignoreDeserializationFailures = false;

   /**
    * Constructor.
    */
   public ExecutionRedisSerializer() {
      super(Execution.class);
      setObjectMapper(objectMapper);
   }

   @Override
   public Execution deserialize(byte[] bytes) throws SerializationException {
      try {
         return super.deserialize(bytes);
      } catch (SerializationException e) {
         if (ignoreDeserializationFailures) {
            log.warn("Ignoring invalid JSON: {}.", e.getMessage());
            return null;
         }

         throw e;
      }
   }

   /**
    * Object mapper.
    */
   public ObjectMapper getObjectMapper() {
      return objectMapper;
   }

   @Override
   public void setObjectMapper(ObjectMapper objectMapper) {
      objectMapper.registerSubtypes(NoResult.class);
      super.setObjectMapper(objectMapper);
      this.objectMapper = objectMapper;
   }

   /**
    * Add additional sub types to the object mapper.
    */
   public void setTypes(Class... types) {
      objectMapper.registerSubtypes(types);
   }

   /**
    * Add additional modules to the object mapper.
    */
   public void setModules(Module... modules) {
      objectMapper.registerModules(modules);
   }

   /**
    * Ignore deserialization failures?.
    */
   public void setIgnoreDeserializationFailures(boolean ignoreDeserializationFailures) {
      this.ignoreDeserializationFailures = ignoreDeserializationFailures;
   }
}

@@ -26,24 +26,127 @@
 rker.json;%0A%0A
+import static java.util.Collections.synchronizedSet;%0A%0Aimport java.util.HashSet;%0Aimport java.util.Set;%0A%0A
 import org.s
@@ -1063,16 +1063,194 @@
 false;%0A%0A
+   /**%0A    * Cache for deserialization failures to reduce log noise.%0A    */%0A   private static final Set%3CString%3E deserializationFailuresCache = synchronizedSet(new HashSet%3C%3E());%0A%0A
    /**%0A 
@@ -1267,16 +1267,16 @@
 ructor.%0A
-
     */%0A 
@@ -1636,63 +1636,277 @@
     
-log.warn(%22Ignoring invalid JSON: %7B%7D.%22, e.getMessage());
+if (deserializationFailuresCache.add(e.getMessage())) %7B%0A               log.warn(%22Ignoring invalid JSON: %7B%7D.%22, e.getMessage());%0A            %7D%0A            if (deserializationFailuresCache.size() %3E 1000) %7B%0A               deserializationFailuresCache.clear();%0A            %7D
 %0A   

Add simple log noise reduction