Al implementar soluciones basadas en Apache Flink que requieren comunicación vía HTTP hacia sus endpoints REST, se enfrenta el desafío de procesar respuestas JSON complejas y extensas. La obtención completa de estos datos requiere una estrategia adecuada de deserialización en el lado del consumidor.

Existen diversas aproximaciones iniciales, cada una con limitaciones significativas:

• Definición manual de DTOs: Mapear manualmente las estructuras JSON genera código propenso a errores y mantenible difícil debido a la profundidad de la anidación.
• Uso de Map genéricos: Aunque flexible, esta opción sacrifica la seguridad de tipos fuertes característica de Java, dificultando el acceso a campos específicos sin casting explícito.
• Dependencia directa de clases fuente: Utilizar las clases nativas de Flink introduce dependencias pesadas y puede generar conflictos de versiones de librerías.

El obstáculo principal radica en cómo Flink maneja sus dependencias internas. Al inspeccionar las clases respuesta, como JobDetailsInfo, se observa que las referencias a librerías externas (como Jackson) están enrutadas a través del esquema sombreado (shaded):

package org.apache.flink.runtime.rest.messages.job;

import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;

public class JobDetailsInfo {
    @JsonProperty("jid")
    private final JobID jobId;
    
    // ... resto de la implementación
}

La causa raíz del fallo al recibir estas entidades reside en la discrepancia entre el paquete donde se definen las anotaciones de serialización y el que espera el mecanismo estándar de conversión de mensajes de Spring. Flink utiliza una copia empaquetada internamente bajo org.apache.flink.shaded.jackson2, mientras que el módulo por defecto de Spring espera encontrar las importaciones estándar de com.fasterxml.jackson.

Estrategia de Solución

Para resolver este conflicto sin alterar el código fuente de Flink ni introducir nuevas dependencias globales conflictivas, se debe construir un convertidor de mensajes HTTP personalizado. Este componente debe heredar la lógica de procesamiento de JSON pero interceptar las importaciones para que apunten al esquema sombreado utilizado por Flink.

Paso 1: Herencia de Clases Base

Se coimenza creando una clase abstracta que extiende la clase base de conversión genérica de Spring. En lugar de utilizar directamente las clases de spring-boot-starter-web, definimos nuestra propia capa intermedia:

public abstract class AbstractShadedJacksonConverter 
    extends AbstractGenericHttpMessageConverter<Object> {
    
    // Lógica base idéntica a AbstractJackson2HttpMessageConverter
    // Sin embargo, todas las dependencias de import cambiarán posteriormente
}

Paso 2: Implementación Específica

A continuación, se implementa la lógica concreta replicando el comportamiento de MappingJackson2HttpMessageConverter. La clave aquí es reemplazar exhaustivamente todos los paquetes de entrada para usar la versión de Jackson incluida en la sombra de Flink. Por ejemplo, en lugar de importar la fábrica de objetos estándar, cargamos la sombra:

import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.http.MediaType;
// Importar otras dependencias de jackson usando el namespace sombreado correspondiente

public class FlinkShadedJsonConverter 
    extends AbstractShadedJacksonConverter {

    private final ObjectMapper objectMapper;

    public FlinkShadedJsonConverter(ObjectMapper mapper) {
        this.objectMapper = mapper;
    }
}

Paso 3: Registro en el Contenedor

Finalmente, es necesario registrar este convertidor dentro de la configuración de Spring. Se configura el objeto ObjectMapper para tolerar propiedades desconocidas y definir zonas horarias adecuadas. También es vital asegurarse de que este componente tenga prioridad sobre cualquier otro convertidor presente, evitando que herramientas alternativas como Fastjson intente procesar el flujo antes que este adaptador:

@Configuration
public class FlinkApiConfiguration {

    @Bean
    public FlinkShadedJsonConverter shadedConverter() {
        ObjectMapper mapper = new ObjectMapper();
        mapper.configure(com.fasterxml.jackson.core.json.JsonWriteFeature.QUOTE_FIELD_NAMES, false);
        mapper.setDateFormat(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"));
        
        return new FlinkShadedJsonConverter(mapper);
    }
}

Es crucial verificar la secuencia de ejecución en el proceso de extracción de datos. Si existen múltiples HttpMessageConverter registrados, el orden determina cuál tiene preferencia. En escenarios donde coexisten convertidores JSON rápidos, el adaptador sombreado debe posicionarse prioritariamente.

Paso 4: Consumo vía Cliente

Con el convertidor habilitado, la interfaz declarativa del cliente (como Feign o WebClient) puede consumir las entidades directamente. No es necesario definir clases manuales intermedias si se confía en la infraestructura configurada:

@Interface
public interface FlinkRestService {
    
    @GetMapping(value = "/v1/jobs/{jobId}", produces = MediaType.APPLICATION_JSON_VALUE)
    ResponseEntity<JobDetailsInfo> getJobInfo(
        @PathVariable("jobId") String jobId, 
        @RequestHeader("X-App-ID") String appId
    );
}