Enabling High Availability

[AlbaFS]

Hello
We are testing up in our dev environment apicurio for kafka.
I would like to ask how to achieve HA runing multiple instances of apicurio in a multibroker, multinode set up.
As well as, where I can find all the configuration properties that can be passed as environment variables (docker-compose) does it support all the kafka properties but in capital letters and underscore?
Thanks!

[EricWittmann]

HA can be a somewhat broad topic. The easiest way to achieve HA for the registry itself is to simply start up multiple instances and point them all to the same Kafka topic.

If you need more than that (multi-broker like you mentioned) then you could use MirrorMaker to duplicate the messages on the apicurio registry Kafka topic into another broker, then point additional apicurio registry instances at that.

It's important to note that a broker failure would result in write operations failing in registry, but read operations would continue to work. So that might be an interesting fact depending on your specific requirements and use-cases.

@carlesarnal can you answer the part about ENV vars for configuring the connection between registry and the kafka broker?

[davidvoit]

Hi Eric,

old discussion but still relevant to us. If we are using the MirrorMaker replicated way, should these replicated nodes be somehow marked readonly, or is there some setting where writes are done on a different kafka topic/cluster then reads (sorry didn't found anything in the documentation and with a quick look in the sourcecode)

Thanks,
David

[AlbaFS]

Many thanks @EricWittmann
So for my understanding, Let's imagine we want three instances of apicurio in different nodes.
I guess that by default they would write to the same topic (kafkasql-journal)
From the consumer and producers part are we allow (quarkus) to point to the three different instances of apicurio having each in a different IP address, or do we have to configure a load balancer round robing them?

Regards.

[carlesarnal]

Hello,

No, not all Kafka properties can be configured right now when using the kafkasql storage. We expose some variables, expose around configuring security and certificates, you can see all of them here.

[carlesarnal]

Many thanks @EricWittmann So for my understanding, Let's imagine we want three instances of apicurio in different nodes. I guess that by default they would write to the same topic (kafkasql-journal) From the consumer and producers part are we allow (quarkus) to point to the three different instances of apicurio having each in a different IP address, or do we have to configure a load balancer round robing them?

Regards.

Getting back to this thanks to the other GH issue, @AlbaFS you're allowed to point each consumer to the replica you need since the data is actually replicated across each instance as long as they use the same kafka topic. That said, the recommended approach is indeed to configure a load balancer in front of the Registry instances to handle the load between them.

As Eric pointed out, if you want to use different Kafka topics then there are more things to take into consideration.

[maver1ck]

@EricWittmann
Hi,
Is there any documentation regarding behaviour of Apicurio when using multiple replicas and Kafka Storage ?
Especially interesting will be information what will happen if one client will try to register same schema in different instances.

Example of this situation:

1. Client send schema registration call to Replica 1. (via LB)
2. Replica 1 register schema and create new record on Kafka topic.
3. Now Replica 1 is killed from any reason (like node failure) before responding to client.
4. Client request timeouts so client send schema registration call to Replica 2.
5. I think now we have race condition situation when we don't know what is first. Replica 2 read from Kafka with new schema or this schema registration call.

Another situation is when we have multiple producents sending (and registering) records with same schema. Example: Flink job with parallelism > 1.

[EricWittmann]

Good question! The semantics here are the same as with a typical SQL database I think, but perhaps with a longer absolute period where problems can occur. What I mean is that a typical REST request to the server would go like this:

1. Client makes REST request (connects to HTTP server)
2. Server processes request into DTO
3. Server connects to DB and writes data
4. Server receives response from DB
5. Server processes DB response into REST object
6. Server writes REST object to HTTP response/socket
7. Client receives response

If the server crashes between 5 and 6 (for example), the client will not receive confirmation of the write, and will assume it failed and try again (typical Kafka app behavior). But it will find that the write was actually done successfully. What to do here?!

I agree that the Kafka storage has the same problem, because it works like this:

1. Client makes REST request (connects to HTTP server)
2. Server processes request
3. Server sends message to Kafka topic
4. Server waits for its own consumer thread to receive the message
5. Consumer thread receives message, updates internal DB
6. Consumer thread notifies HTTP thread
7. Server processes information from consumer thread notification into REST object
8. Server writes REST object to HTTP response/socket
9. Client receives response

The "storage" is officially updated as soon as (3) is successful (the message is added to the Kafka topic). So if the replica dies after (3), the client will think the operation failed when it did not.

IMO the solution/mitigation to both of these is to use the "Create Artifact" endpoint with the ifExists parameter set appropriately:

In particular the RETURN_OR_UPDATE option is pretty useful here. Alternatively the client could try the classic pattern of:

1. Check if schema exists already
2. If not, register it

Or the equally classic

1. Try to register schema
2. If failure, query for existing schema

I would probably go with the ifExists approach. :)

[EricWittmann]

Note: if the client retries the request so quickly that replica 2 receives the request before it has processed the message sent to the Kafka topic by replica 1, then a number of possible things could happen:

a) If the request is a read query, old data might be returned
b) If the request is a conflicting write, it will fail
c) If the request is a hybrid (e.g. create request with ifExists set to UPDATE) it will work as though replica 1 didn't fail

Why is this?

(a) Reads are executed in parallel in the Kafka storage implementation. It is possible for a write sent to replica 1 to succeed and return a REST response and then after that send a read request to replica 2 and get back stale information. This is unlikely but possible because all replicas are consuming the messages on the Kafka topic independently. So replica 2 might be a little bit slower than replica 1, for example. Which would result in potentially stale answers to read queries.

(b) Writes are executed in sequence in the Kafka storage implementation. The order in which writes are processed is dictated by the order of the messages in the Kafka topic. We have chosen our partition keys specifically to rely on this ordering. So for write operations if replica 1 dies and the client re-sends the request, replica 2 is guaranteed to process the write operation from replica 1 before processing its own write operation (the re-sent message).

(c) See explanation for (b). :) In the case of ifExists set to e.g. RETURN_OR_UPDATE that operation is effectively a write. It attempts to perform the write operation first, and then falls back to some behavior if that write fails (which in the scenario above it will fail because the message from replica 1 would exist in the Kafka topic).

[EricWittmann]

We don't have formalized documentation explaining the implementation of the Kafka storage, but I hope that helps. I would expect the Flink job with parallelism > 1 should work just fine if you use the ifExists approach, for the reasons explained above. This is exactly what we recommend for Kafka architectures with multiple identical producer applications. They should just try to create the schema when they start up and set ifExists to RETURN_OR_UPDATE. Let the server figure it out -

A) Create and return the artifact if this is the first time and the first version
B) Create a new version of the artifact if this is the first time but one or more older version already exists
C) Return the existing version if this is not the first time

Note that the server uses the hash of the content being registered to determine if a version already exists. You can use a canonical hash (more forgiving) or a standard hash (brutal/strict).

[maver1ck]

@EricWittmann,
Your answer is very helpful and includes all the implementation details needed to understand Apicurio behaviour.
Thank you.

I think it should be included in official documentation but eventually Google will index it here :)