It crashes when the dict is converted to an swh-model object.

Related to D2813
Related to D2820

This allows to decrease the number of contents not replayed when hash
collisions happened.

What we really want is for the broker to acknowledge all messages before we go
on to the next step. That's accomplished by flushing the producer rather than
enabling idempotence (which has other side-effects, such as only-once delivery,
which we don't really care about as all our consumers are, in effect,
idempotent).

Setting acks to all means that the broker acknowledges that all in-sync replicas
have persisted the message, which is a stronger guarantee than what we had
before.

This setting needs to be tuned differently for different topics, as the size of
objects varies quite wildly. Bumping the default to 200 reduces chatter in the
consumer group which should reduce the amount of bandwidth used for consumption.

This paves the way for removal of all the deprecated cli options in `swh
journal`.

 - Drop catch_exception as it's not used anywhere (and we always assert on
 result.return_code)
 - Make the argument list a starargs instead of having to make a list

In preparation of being able to override parts of the configuration.

remove the 'flush' argument from these methods' argument list: it
does not exists in the InMemoryJournalWriter version of this
service, and it is in fact useless.

After various refactorings, the meaning of `max_messages` got muddled and we're
at a point where it doesn't mean anything anymore. `stop_after_objects` is
clearer as to what behavior the parameter is actually trying to achieve. We also
rename the replay command-line argument to the same name.

These refactorings also had us end up with a loop inside JournalClient.process,
while some callers still had a loop around it which ever got called once (or 0
times when the surrounding code was buggy). This commit removes all these outer
loops as well, keeping only the `JournalClient.process` inner loop.

While we're here, we use the opportunity to clarify and expand the documentation
of the JournalClient.

When using the object_dicts() strategy, we often end up not generating any
contents; This works because the `replayer.process` call is within a loop,
guarded with the length of the queue.

If the queue is empty, `replayer.process` is never called, but the test is also
useless. So we add an assertion to that effect.

This allows getting more verbose logs from kafka during tests on packaged
versions of the module (e.g. when using tox).

This uses a custom wrapper exception and tenacity callbacks to log exceptions
when the copy of a given content fails several times. This makes the consumer
more robust (fewer crashes), which in turns allows fewer consumer rebalances,
which finally drastically reduces the consumer bandwidth consumption.

At this point, the retry of "definitely" failed content replays needs to be
handled separately.

This will allow us to reuse it in more tests.

Uses logging keyword arguments for filtering, instead of plain arguments

Containment checking has a tendency to fail with an error 500 on S3, retrying
smooths that out.

Replaces docstring annotations with type annotations as well

This avoids reusing the same consumer group name on subsequent tests, which is a
problem when some of the tests change the broker-side consumer group rebalance
timeouts.

This trades bandwidth/processing time for more API queries, which can be a win
if your exclusion file is a bit stale.

This allows us to get a better sense of the actual work done by the content
replayer.

The new KAFKA_GROUP_INSTANCE_ID env variable can be set on journal clients to
set the group.instance.id enable support for Kafka Static Consumer Groups.

https://github.com/edenhill/librdkafka/pull/2525
https://cwiki.apache.org/confluence/display/KAFKA/KIP-345%3A+Introduce+static+membership+protocol+to+reduce+consumer+rebalances

Combined with larger values for the session.timeout.ms and max.poll.interval.ms
settings, this setting informs the Group Coordinator broker that the consumer
group has static membership, and that the disappearance of a given member of the
consumer group should not immediately trigger a rebalance; This allows crashing
consumers to re-join the consumer group and start consuming from their assigned
partitions immediately.

This setting is implemented as an environment variable so that several consumers
in the same group can share a configuration file, and still override the
value (e.g. by setting `Environment=KAFKA_GROUP_INSTANCE_ID=groupname-%i` in a
systemd template unit).

When this setting is enabled, we also up the relevant values for
session.timeout.ms and max.poll.interval.ms.

This matches the production environment.

We need to add some bits to the zookeeper configuration, as the zookeeper
bundled with kafka 2.4.0 starts the adminserver, which tries to bind to port
8080, by default.

It makes the consumer crash with `generator didn't stop after throw()`
exceptions. We'll reintroduce retry behavior at a later stage.

This change introduced subtly confusing behavior, where the storage.objstorage
__getattr__ proxy hides itself, until you try a dunder method (e.g. `foo in
storage.objstorage` or `iter(storage.objstorage)`), blows up in your face with a
confusing exception (because the collaborator is _also_ called ObjStorage).

Using the underlying objstorage works around that issue.

This fixes the current ci build failure [1]

[1] https://jenkins.softwareheritage.org/job/DJNL/job/tests/717/console

Allow clients to override this behavior more easily in the client,
notably the deserialization step where message metadata can be added to
the objects.

Required by swh-storage >= v0.0.172.

This new endpoint is now required by swh-storage for skipped contents.