diff --git a/rele/subscription.py b/rele/subscription.py
index 846618d..73f6b03 100644
--- a/rele/subscription.py
+++ b/rele/subscription.py
@@ -1,7 +1,6 @@
 import json
 import logging
 import time
-from concurrent import futures
 
 from .middleware import run_middleware_hook
 
@@ -64,7 +63,19 @@ class Callback:
         run_middleware_hook("pre_process_message", self._subscription, message)
         start_time = time.time()
 
-        data = json.loads(message.data.decode("utf-8"))
+        try:
+            data = json.loads(message.data.decode("utf-8"))
+        except json.JSONDecodeError as e:
+            message.ack()
+            run_middleware_hook(
+                "post_process_message_failure",
+                self._subscription,
+                e,
+                start_time,
+                message,
+            )
+            run_middleware_hook("post_process_message")
+            return
 
         try:
             res = self._subscription(data, **dict(message.attributes))
diff --git a/rele/worker.py b/rele/worker.py
index 08ea6fa..6b1e901 100644
--- a/rele/worker.py
+++ b/rele/worker.py
@@ -3,6 +3,8 @@ import sys
 import time
 from concurrent import futures
 
+from google.cloud.pubsub_v1.subscriber.scheduler import ThreadScheduler
+
 from .client import Subscriber
 from .middleware import run_middleware_hook
 from .subscription import Callback
@@ -56,9 +58,10 @@ class Worker:
             executor_kwargs = {
                 "thread_name_prefix": "ThreadPoolExecutor-ThreadScheduler"
             }
-            scheduler = futures.ThreadPoolExecutor(
+            executor = futures.ThreadPoolExecutor(
                 max_workers=self.threads_per_subscription, **executor_kwargs
             )
+            scheduler = ThreadScheduler(executor=executor)
             self._futures.append(
                 self._subscriber.consume(
                     subscription_name=subscription.name,

'ThreadPoolExecutor' object has no attribute 'queue'
On Relé 0.9.0 I get the following error when starting a worker:

```
Starting Rele
Traceback (most recent call last):
  File "worker.py", line 12, in <module>
    worker.run_forever()
  File "/usr/local/lib/python3.7/site-packages/rele/worker.py", line 77, in run_forever
    self.start()
  File "/usr/local/lib/python3.7/site-packages/rele/worker.py", line 66, in start
    scheduler=scheduler,
  File "/usr/local/lib/python3.7/site-packages/rele/client.py", line 88, in consume
    subscription_path, callback=callback, scheduler=scheduler
  File "/usr/local/lib/python3.7/site-packages/google/cloud/pubsub_v1/subscriber/client.py", line 228, in subscribe
    manager.open(callback=callback, on_callback_error=future.set_exception)
  File "/usr/local/lib/python3.7/site-packages/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py", line 429, in open
    self._dispatcher = dispatcher.Dispatcher(self, self._scheduler.queue)
AttributeError: 'ThreadPoolExecutor' object has no attribute 'queue'
Sentry is attempting to send 0 pending error messages
Waiting up to 2 seconds
Press Ctrl-C to quit
```

The issue is that we are passing an executor to the [consume API](https://github.com/mercadona/rele/blob/master/rele/worker.py#L66), when in fact it should be an[ instance of a Scheduler](https://googleapis.dev/python/pubsub/1.0.2/subscriber/api/scheduler.html#google.cloud.pubsub_v1.subscriber.scheduler.ThreadScheduler)
diff --git a/tests/test_subscription.py b/tests/test_subscription.py
index ec33e67..d3e07d5 100644
--- a/tests/test_subscription.py
+++ b/tests/test_subscription.py
@@ -91,14 +91,33 @@ class TestCallback:
             message_id="1",
         )
 
-        return pubsub_v1.subscriber.message.Message(rele_message, "ack-id", MagicMock())
+        message = pubsub_v1.subscriber.message.Message(
+            rele_message, "ack-id", MagicMock()
+        )
+        message.ack = MagicMock(autospec=True)
+        return message
 
     @pytest.fixture
     def message_wrapper_empty(self):
         rele_message = pubsub_v1.types.PubsubMessage(
             data=b"", attributes={"lang": "es"}, message_id="1"
         )
-        return pubsub_v1.subscriber.message.Message(rele_message, "ack-id", MagicMock())
+        message = pubsub_v1.subscriber.message.Message(
+            rele_message, "ack-id", MagicMock()
+        )
+        message.ack = MagicMock(autospec=True)
+        return message
+
+    @pytest.fixture
+    def message_wrapper_invalid_json(self):
+        rele_message = pubsub_v1.types.PubsubMessage(
+            data=b"foobar", attributes={}, message_id="1"
+        )
+        message = pubsub_v1.subscriber.message.Message(
+            rele_message, "ack-id", MagicMock()
+        )
+        message.ack = MagicMock(autospec=True)
+        return message
 
     def test_log_start_processing_when_callback_called(
         self, caplog, message_wrapper, published_at
@@ -129,6 +148,7 @@ class TestCallback:
             res = callback(message_wrapper)
 
         assert res == 123
+        message_wrapper.ack.assert_called_once()
         assert len(caplog.records) == 3
         message_wrapper_log = caplog.records[1]
         assert message_wrapper_log.message == (
@@ -168,6 +188,7 @@ class TestCallback:
         res = callback(message_wrapper)
 
         assert res is None
+        message_wrapper.ack.assert_not_called()
         failed_log = caplog.records[-1]
         assert failed_log.message == (
             "Exception raised while processing "
@@ -187,6 +208,33 @@ class TestCallback:
         }
         assert failed_log.subscription_message == message_wrapper
 
+    def test_log_acks_called_message_when_not_json_serializable(
+        self, caplog, message_wrapper_invalid_json, published_at
+    ):
+        callback = Callback(sub_stub)
+        res = callback(message_wrapper_invalid_json)
+
+        assert res is None
+        message_wrapper_invalid_json.ack.assert_called_once()
+        failed_log = caplog.records[-1]
+        assert failed_log.message == (
+            "Exception raised while processing "
+            "message for rele-some-cool-topic - "
+            "sub_stub: JSONDecodeError"
+        )
+        assert failed_log.metrics == {
+            "name": "subscriptions",
+            "data": {
+                "agent": "rele",
+                "topic": "some-cool-topic",
+                "status": "failed",
+                "subscription": "rele-some-cool-topic",
+                "duration_seconds": pytest.approx(0.5, abs=0.5),
+                "attributes": {},
+            },
+        }
+        assert failed_log.subscription_message == message_wrapper_invalid_json
+
     def test_published_time_as_message_attribute(self, message_wrapper, caplog):
         callback = Callback(sub_published_time_type)
         callback(message_wrapper)
diff --git a/tests/test_worker.py b/tests/test_worker.py
index f8fb506..9d3d41b 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -3,6 +3,7 @@ from unittest.mock import ANY, patch
 import os
 
 import pytest
+from google.cloud.pubsub_v1.subscriber.scheduler import ThreadScheduler
 
 from rele import Subscriber, Worker, sub
 from rele.middleware import register_middleware
@@ -47,7 +48,8 @@ class TestWorker:
             subscription_name="rele-some-cool-topic", callback=ANY, scheduler=ANY
         )
         scheduler = mock_consume.call_args_list[0][1]["scheduler"]
-        assert isinstance(scheduler, futures.ThreadPoolExecutor)
+        assert isinstance(scheduler, ThreadScheduler)
+        assert isinstance(scheduler._executor, futures.ThreadPoolExecutor)
 
     def test_setup_creates_subscription_when_topic_given(
         self, mock_create_subscription, worker
@@ -71,7 +73,8 @@ class TestWorker:
             subscription_name="rele-some-cool-topic", callback=ANY, scheduler=ANY
         )
         scheduler = mock_consume.call_args_list[0][1]["scheduler"]
-        assert isinstance(scheduler, futures.ThreadPoolExecutor)
+        assert isinstance(scheduler, ThreadScheduler)
+        assert isinstance(scheduler._executor, futures.ThreadPoolExecutor)
         mock_wait_forever.assert_called_once()
 
     @patch.object(Worker, "_wait_forever")

pip install -e .[dev]
parse_log_pytest
requirements.txt
pytest pytest-cov pytest-xdist pytest-mock pytest-asyncio black pytest
apt-get update apt-get install -y gcc
3.8
requirements/base.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
black==24.8.0
cachetools==5.5.2
certifi==2025.1.31
charset-normalizer==3.4.1
click==8.1.8
coverage==7.6.1
Deprecated==1.2.18
exceptiongroup==1.2.2
execnet==2.1.1
google-api-core==2.24.2
google-auth==2.38.0
google-cloud-pubsub==2.29.0
googleapis-common-protos==1.69.2
grpc-google-iam-v1==0.14.2
grpcio==1.70.0
grpcio-status==1.70.0
idna==3.10
importlib_metadata==8.5.0
iniconfig==2.1.0
mypy-extensions==1.0.0
opentelemetry-api==1.31.1
opentelemetry-sdk==1.31.1
opentelemetry-semantic-conventions==0.52b1
packaging==24.2
pathspec==0.12.1
platformdirs==4.3.6
pluggy==1.5.0
proto-plus==1.26.1
protobuf==5.29.4
pyasn1==0.6.1
pyasn1_modules==0.4.2
pytest==8.3.5
pytest-asyncio==0.24.0
pytest-cov==5.0.0
pytest-mock==3.14.0
pytest-xdist==3.6.1
-e git+https://github.com/mercadona/rele.git@56966a34d6e12209eb81f6b672dbe1bc1aa14278#egg=rele
requests==2.32.3
rsa==4.9
tomli==2.2.1
typing_extensions==4.13.0
urllib3==2.2.3
wrapt==1.17.2
zipp==3.20.2

name: rele
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=3.0.16=h5eee18b_0
  - pip=24.2=py38h06a4308_0
  - python=3.8.20=he870216_0
  - readline=8.2=h5eee18b_0
  - setuptools=75.1.0=py38h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - wheel=0.44.0=py38h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - black==24.8.0
      - cachetools==5.5.2
      - certifi==2025.1.31
      - charset-normalizer==3.4.1
      - click==8.1.8
      - coverage==7.6.1
      - deprecated==1.2.18
      - exceptiongroup==1.2.2
      - execnet==2.1.1
      - google-api-core==2.24.2
      - google-auth==2.38.0
      - google-cloud-pubsub==2.29.0
      - googleapis-common-protos==1.69.2
      - grpc-google-iam-v1==0.14.2
      - grpcio==1.70.0
      - grpcio-status==1.70.0
      - idna==3.10
      - importlib-metadata==8.5.0
      - iniconfig==2.1.0
      - mypy-extensions==1.0.0
      - opentelemetry-api==1.31.1
      - opentelemetry-sdk==1.31.1
      - opentelemetry-semantic-conventions==0.52b1
      - packaging==24.2
      - pathspec==0.12.1
      - platformdirs==4.3.6
      - pluggy==1.5.0
      - proto-plus==1.26.1
      - protobuf==5.29.4
      - pyasn1==0.6.1
      - pyasn1-modules==0.4.2
      - pytest==8.3.5
      - pytest-asyncio==0.24.0
      - pytest-cov==5.0.0
      - pytest-mock==3.14.0
      - pytest-xdist==3.6.1
      - requests==2.32.3
      - rsa==4.9
      - tomli==2.2.1
      - typing-extensions==4.13.0
      - urllib3==2.2.3
      - wrapt==1.17.2
      - zipp==3.20.2
prefix: /opt/conda/envs/rele

tests/test_worker.py::TestWorker::test_start_subscribes_and_saves_futures_when_subscriptions_given tests/test_worker.py::TestWorker::test_run_sets_up_and_creates_subscriptions_when_called
tests/test_subscription.py::TestSubscription::test_executes_callback_when_called tests/test_worker.py::TestWorker::test_stop_closes_db_connections
tests/test_subscription.py::TestSubscription::test_subs_return_subscription_objects tests/test_subscription.py::TestSubscription::test_subs_without_prefix_return_subscription_objects tests/test_subscription.py::TestSubscription::test_sub_executes_when_message_attributes_match_criteria tests/test_subscription.py::TestSubscription::test_sub_does_not_execute_when_message_attributes_dont_match_criteria tests/test_worker.py::TestWorker::test_setup_creates_subscription_when_topic_given tests/test_worker.py::TestWorker::test_wait_forevers_for_custom_time_period_when_called_with_argument tests/test_worker.py::TestWorker::test_creates_subscription_with_custom_ack_deadline_from_environment tests/test_worker.py::TestWorker::test_creates_without_config
rele/worker.py rele/subscription.py
rele/worker.py rele/subscription.py