@SuppressWarnings("unchecked")
    protected <T extends HousekeeperTask> void processTask(ToHousekeeperServiceMsg msg) throws Exception {
        HousekeeperTask task = JacksonUtil.fromString(msg.getTask().getValue(), HousekeeperTask.class);
        HousekeeperTaskType taskType = task.getTaskType();
        if (config.getDisabledTaskTypes().contains(taskType)) {
            log.debug("Task type {} is disabled, ignoring {}", taskType, task);
            return;
        }
        HousekeeperTaskProcessor<T> taskProcessor = (HousekeeperTaskProcessor<T>) taskProcessors.get(taskType);
        if (taskProcessor == null) {
            throw new IllegalArgumentException("Unsupported task type " + taskType);
        }

        try {
            long startTs = System.currentTimeMillis();
            Future<Object> future = taskExecutor.submit(() -> {
                taskProcessor.process((T) task);
                return null;
            });
            future.get(config.getTaskProcessingTimeout(), TimeUnit.MILLISECONDS);

            long timing = System.currentTimeMillis() - startTs;
            if (log.isDebugEnabled()) {
                log.debug("[{}] Processed {} in {} ms (attempt {})", task.getTenantId(), task.getDescription(), timing, msg.getTask().getAttempt());
            }
            statsService.ifPresent(statsService -> statsService.reportProcessed(taskType, msg, timing));
        } catch (InterruptedException e) {
            throw e;
        } catch (Throwable e) {
            Throwable error = e;
            if (e instanceof ExecutionException) {
                error = e.getCause();
            } else if (e instanceof TimeoutException) {
                error = new TimeoutException("Timeout after " + config.getTaskProcessingTimeout() + " seconds");
            }

            if (msg.getTask().getAttempt() < config.getMaxReprocessingAttempts()) {
                log.warn("[{}] Failed to process {} (attempt {}), submitting for reprocessing",
                        task.getTenantId(), task.getDescription(), msg.getTask().getAttempt(), error);
                reprocessingService.submitForReprocessing(msg, error);
            } else {
                log.error("[{}] Failed to process task in {} attempts: {}", task.getTenantId(), msg.getTask().getAttempt(), msg, e);
                notificationRuleProcessor.process(TaskProcessingFailureTrigger.builder()
                        .task(task)
                        .error(error)
                        .attempt(msg.getTask().getAttempt())
                        .build());
            }
            statsService.ifPresent(statsService -> statsService.reportFailure(taskType, msg));
        }
    }
@Test
    public void whenReprocessingAttemptsExceeded_thenDropTheTask() throws Exception {
        TimeoutException error = new TimeoutException("Test timeout");
        doThrow(error).when(tsHistoryDeletionTaskProcessor).process(any());

        Device device = createDevice("test", "test");
        createRelatedData(device.getId());

        doDelete("/api/device/" + device.getId()).andExpect(status().isOk());

        int maxAttempts = 5;
        await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> {
            for (int i = 1; i <= maxAttempts; i++) {
                verifyTaskProcessing(device.getId(), HousekeeperTaskType.DELETE_TS_HISTORY, i);
            }
        });

        Mockito.clearInvocations(housekeeperService);
        doCallRealMethod().when(tsHistoryDeletionTaskProcessor).process(any());
        TimeUnit.SECONDS.sleep(2);
        verify(housekeeperService, never()).processTask(argThat(getTaskMatcher(device.getId(), HousekeeperTaskType.DELETE_TS_HISTORY, null)));
    }