David Douard
--- a/swh/journal/pytest_plugin.py
+++ b/swh/journal/pytest_plugin.py

        fetched_messages += 1
        topic = msg.topic()
-        assert topic.startswith(kafka_prefix + "."), "Unexpected topic"
+        assert topic.startswith(f"{kafka_prefix}.") or topic.startswith(
+            f"{kafka_prefix}_privileged."
--- a/swh/journal/tests/test_client.py
+++ b/swh/journal/tests/test_client.py
            stop_after_objects=1,
            object_types=["else"],
        )
+
+
+def test_client_subscriptions_with_anonymized_topics(
+    kafka_prefix: str, kafka_consumer_group: str, kafka_server_base: str
+):
+    producer = Producer(
--- a/swh/journal/tests/test_client.py
+++ b/swh/journal/tests/test_client.py
+        prefix=kafka_prefix,
+        stop_after_objects=1,
+        privileged=False,
+    )
+    # we only subscribed to the standard prefix
+    assert client.subscription == [kafka_prefix + ".revision"]
+
+    # with privileged channel activated on the client side
+    client = JournalClient(
+        brokers=[kafka_server_base],
+        group_id=kafka_consumer_group,
+        prefix=kafka_prefix,
+        stop_after_objects=1,
+        privileged=True,
+    )
+    # we also only subscribed to the standard prefix, since there is no priviled prefix
--- a/swh/journal/tests/test_kafka_writer.py
+++ b/swh/journal/tests/test_kafka_writer.py
    consumed_messages = consume_messages(consumer, kafka_prefix, expected_messages)
    assert_all_objects_consumed(consumed_messages)

+    for key, obj_dict in consumed_messages["revision"]:
+        obj = Revision.from_dict(obj_dict)
+        for person in (obj.author, obj.committer):
+            assert not (
+                len(person.fullname) == 32
+                and person.name is None
+                and person.email is None
--- a/swh/journal/writer/kafka.py
+++ b/swh/journal/writer/kafka.py

    def _write_addition(self, object_type: str, object_: ModelObject) -> None:
        """Write a single object to the journal"""
-        topic = f"{self._prefix}.{object_type}"
        key = object_key(object_type, object_)
+
+        if self.anonymize:
+            anon_object_ = object_.anonymize()
+            if anon_object_:  # can be either None, or an anonymized object
+                # if the object is anonymizable, send the non-anonymized version in the
+                # privileged channel
+                topic = f"{self._prefix_privileged}.{object_type}"
+                dict_ = self._sanitize_object(object_type, object_)
+                logger.debug("topic: %s, key: %s, value: %s", topic, key, dict_)
+                self.send(topic, key=key, value=dict_)