Adding/removing physical segment for a logical table

[erlis1]

I have the following setup:

2 (two) pinot servers hosting same real-time table (same table name / kafka consumer group id); each one of the servers receivs 1/2 of total of msgs from the kafka stream.
1 (one zookeper), 1 (one) broker and 1 (one) controller

What happens if one of the servers crashes (the server process itself), I understand that the second server will consume 100% of the messages while the second server is down (I can see that the broker query returns about half of the documets numbers in this case), but how can I add the second server again (if at all) in order to restore the original setup ?

What's going to happen if a broker/controller crashes ?

In general, how should I manage adding/removing pinot segments for reducing workload balance and/or system crash ?

Thank you

[jfim]

Currently, the realtime consumption in Pinot uses the high level consumer which leads to the behavior you're seeing.

If you want to create two servers that consume from Kafka, each of them should get their own consumer group ID so that they each get all of the data; since the realtime segment boundaries include offsets, all replicas of the realtime segments should contain the same data, so a server node that crashes does not change the query results.

This is not ideal and it's on our roadmap to improve Kafka consumption.

[kishoreg]

What happens if one of the servers crashes (the server process itself), I
understand that the second server will consume 100% of the messages while
the second server is down (I can see that the broker query returns about
half of the documets numbers in this case), but how can I add the second
server again (if at all) in order to restore the original setup ?

There are two scenarios,

#1. adding a new server

In this case, the new server will start handling 1/2 of the total msgs. I
think, there is a need to tag the new machine with the same name as the old
one. I know there is an api for this buts its probably not documented. (we
will provide a command to do this).

#.Bringing the failed server back up

In this case, the messages will again be distributed equally among the two
nodes. There is no need to make any additional modifications to the tag as
in previous case.

In general, you want to use have multiple replicas to handle failures.

Broker: Have more than one broker will provide you fault tolerance. The
request can be forwarded to any live broker.

Controller: Simply start multiple controllers, there is leader election
among the controllers and only one controller will manage the cluster. If
the leader fails, a new leader is elected and it starts managing the
cluster.

Server: Here when you create a resource, you specify the number of replicas
needed for each segment. Typically we use 3 replicas in our deployment. In
case of failure, broker automatically detects the failures and routes the
queries to live servers.

Hope this helps.

Thanks,
Kishore G

On Thu, Jun 25, 2015 at 4:02 PM, Jean-François Im notifications@github.com
wrote:

Currently, the realtime consumption in Pinot uses the high level consumer
which leads to the behavior you're seeing.

If you want to create two servers that consume from Kafka, each of them
should get their own consumer group ID so that they each get all of the
data; since the realtime segment boundaries include offsets, all replicas
of the realtime segments should contain the same data, so a server node
that crashes does not change the query results.

This is not ideal and it's on our roadmap to improve Kafka consumption.

—
Reply to this email directly or view it on GitHub
https://github.com/linkedin/pinot/issues/16#issuecomment-115424476.

[erl1s]

Thank you for your answers,
Can you please be more specific about the how server vs replicas work?
I'm taking the scenario of real-time consumption and storing same table on multiple servers.

Are replicas created and queried by the server process or using the controllers ?
I'm asking that because I want to be sure that I know how to deal with potential failures.

A server process dumps its segment data to the hard disk (I assume that we use 3 replicas per segment), and this data is read when query is received.
If suddenly the server process stopped (due to an exception memory error etc.) - the stored data of the replicas isn't reachable anymore, right ? It's not ever relevant that the data is replicated, because the process which is responsible for reading the data is down.
Let's say I re-start the same failed server process again. Will the old replicated data will be available as soon as the new server process is up ? Will it continue to add new data to the old replicas, or will create new replicas and the old (pre-failure) data will be lost?

Thank you again. Pinot is an exciting project and I really want to understand how to use it in my case.

P.S. if there's some technologies that Pinot is build on and will answer my questions I'd be happy to read about it.