Aruha 456 block consumers #489

v-stepanov · 2016-11-22T11:37:38Z

@antban @rcillo @adyach @v1ctor @SergKam please review
(It was not yet tested properly, i will create a separate stack and will test it there, so if you notice some bug in the code - let me know :)

I have no idea how this small feature could grow to 700 lines on code :(

Concept document moved to wiki: https://github.com/zalando/nakadi/wiki/Too-much-consumers-from-a-single-application

… consuming connections;

codecov-io · 2016-11-22T12:09:23Z

Current coverage is 59.07% (diff: 14.41%)

Merging #489 into master will decrease coverage by 2.29%

@@             master       #489   diff @@
==========================================
  Files           159        164     +5   
  Lines          4279       4489   +210   
  Methods           0          0          
  Messages          0          0          
  Branches        481        500    +19   
==========================================
+ Hits           2626       2652    +26   
- Misses         1488       1670   +182   
- Partials        165        167     +2

Powered by Codecov. Last update 17ebb2a...cc44af0

adyach · 2016-11-22T12:49:50Z

@v-stepanov are you going to move concept document to wiki and post the link here?

adyach · 2016-11-22T13:03:49Z

src/main/java/org/zalando/nakadi/service/ConsumerLimitingService.java

+import java.util.stream.Collectors;
+
+import static java.text.MessageFormat.format;
+import static org.echocat.jomon.runtime.concurrent.Retryer.executeWithRetry;


Didn't we agree on not to use static imports ? :)

Hell no! :)
We had a balance of -1 in our radar which doesn't allow neither adopt nor forbid that.

adyach · 2016-11-22T13:15:57Z

src/main/java/org/zalando/nakadi/service/ConsumerLimitingService.java

+            throws ConnectionSlotOccupiedException {
+
+        final String parent = zkPathForConsumer(client, eventType, partition);
+        List<String> children = ImmutableList.of();


Maybe we can init it only in case of KeeperException.NoNodeException ?

Good point. Fixed 37f2d31

adyach · 2016-11-22T13:28:21Z

src/main/java/org/zalando/nakadi/service/ConsumerLimitingService.java

+                .filter(slot -> !occupiedSlots.contains(slot))
+                .collect(Collectors.toList());
+        final int slotIndex = new Random().nextInt(availableSlots.size());
+        final String slot = availableSlots.get(slotIndex);


The issue with this approach that the change of nakadi.stream.maxConnections to value grater than 5 will cause an exception.

Very good catch. I fixed it dedca7c

adyach · 2016-11-22T13:30:57Z

src/main/java/org/zalando/nakadi/service/ConsumerLimitingService.java

+            for (final String partition : partitions) {
+                Optional<ConnectionSlot> connectionSlot = Optional.empty();
+                try {
+                    connectionSlot = executeWithRetry(


for me it is not clear why should we retry here

As you see the implementation doesn't use locks - because that puts additional load on ZK and also lock nodes are not cleaned after. So to avoid of creation of more than 5 connections the implementation uses named slots ("1", "2", "3", "4", "5"). And the possible situation is that two instances will try to capture the same free slot at the same time. In that case we should retry to try capture another free slot.
(Maybe not the best explanation but feel free to approach me so that I can describe how it works on board)

v-stepanov · 2016-11-23T07:07:07Z

@adyach I moved now the concept document to project wiki: https://github.com/zalando/nakadi/wiki/Too-much-consumers-from-a-single-application

rcillo · 2016-11-23T10:28:07Z

src/acceptance-test/java/org/zalando/nakadi/webservice/EventStreamReadingAT.java

@@ -327,6 +334,52 @@ private Response readEvents() {
    }

    @Test(timeout = 10000)
+    public void whenExceedMaxConsumersNumThen429() throws IOException, InterruptedException, ExecutionException,


@v-stepanov it would be good to have a test case for when a connection is closed. It should release one slot.

I extended this test. Now it also checks if one slot is released when connection is closed. c8f0987

rcillo · 2016-11-23T10:39:36Z

src/main/java/org/zalando/nakadi/service/ConsumerLimitingService.java

+
+    public static final String CONNECTIONS_ZK_PATH = "/nakadi/consumers/connections";
+
+    private static final Logger LOG = LoggerFactory.getLogger(CursorsService.class);


s/CursorsService/ConsumerLimitingService/

Good point. I fixed that now d82978a

rcillo · 2016-11-23T12:18:59Z

src/main/java/org/zalando/nakadi/service/ConsumerLimitingService.java

+                try {
+                    connectionSlot = executeWithRetry(
+                            () -> acquireConnectionSlot(client, eventType, partition),
+                            new RetryForSpecifiedCountStrategy<Optional<ConnectionSlot>>(5)


The number of retries should be at least maxConnections and not a fixed 5.

If I understood correctly, slots number starts at 0 and goes up to maxConnections . If there is concurrency when getting 0, the next attempt will be to get 1, and so on. So if in the future we increase maxConnections to 10, for example, this algorithm would stop trying on the fifith slot and reply with 429 even though there were another 5 slots.

Makes sense?

It's not exactly like you described. The client tries to capture one of FREE slots (so he doesn't try blindly), and also he chooses a random free slot (not the lowest one). So the retry will happen only if two clients were trying to capture the same free slot at the same time.
But anyway I see your point. Will change the number of retries to maxConnection.

changed fde52a9

rcillo · 2016-11-23T12:22:47Z

👍

antban · 2016-11-23T12:55:01Z

src/main/java/org/zalando/nakadi/service/ConsumerLimitingService.java

+        }
+    }
+
+    public void deletePartitionNodeIfPossible(final String nodePath) {


Well, this method can be probably renamed to deleteAnyZkNode

rcillo · 2016-11-30T14:44:18Z

I see no more pending comments.

rcillo · 2016-11-30T14:44:24Z

…sumers

adyach · 2016-12-01T07:53:16Z

src/main/java/org/zalando/nakadi/service/ConsumerLimitingService.java

+
+    private List<String> getChildrenCached(final String zkPath) {
+        try {
+            PathChildrenCache cache = SLOTS_CACHES.getOrDefault(zkPath, null);


Why don't you want to use curator TreeCache for that? In that case you won't store the list of caches.

I don't really need to cache all the nodes' children in "/nakadi/consumers/connections"
I'm only interested in nodes's children for connections that exist in this Nakadi instance.
So the TreeCache is too much for me in this case, it will create a huge amount of watchers which I really don't need and that will put a very high load on ZK.

adyach · 2016-12-01T07:56:37Z

src/main/java/org/zalando/nakadi/service/ConsumerLimitingService.java

+            if (cache == null) {
+                cache = new PathChildrenCache(zkHolder.get(), zkPath, false);
+                cache.start(BUILD_INITIAL_CACHE);
+                SLOTS_CACHES.put(zkPath, cache);


In case of two threads are in this if block It is possible that you put one cache into the map then other thread get it in line 205 but after another thread in if block will replace it with new object.
I suggest to use putIfAbsent or even replace SLOTS_CACHES.getOrDefault(zkPath, null) with computeIfAbsent.

Very good point! Thanks. I fixed it now the way you suggested f41ca10

adyach · 2016-12-01T08:16:11Z

src/main/java/org/zalando/nakadi/service/ConsumerLimitingService.java

+                        && s.getClient().equals(slot.getClient())
+                        && s.getEventType().equals(slot.getEventType()));
+        if (!hasMoreConnectionsToPartition) {
+            final String consumerPath = zkPathForConsumer(slot.getClient(), slot.getEventType(), slot.getPartition());


For example the thread came to this if block and the same moment another thread will acquire connection for that same slot. The node exists in the ZK but not in the ACQUIRED_SLOTS . Do you think it is gonna be a problem?

I think I don't completely get it.

The node exists in the ZK but not in the ACQUIRED_SLOTS

Why you think it will not be in ACQUIRED_SLOTS ?

In general, I did not get why you need ACQUIRED_SLOTS. You can get acquired slots from cache because ACQUIRED_SLOTS do not give you any synchronisation between cache and ZK.

The only reason why i need ACQUIRED_SLOTS is to be able to delete the cache which I don't need any more at current Nakadi instance. Without knowing what are the current acquired slots at this instance I will not know if I can delete the cache already or if I still need it.

adyach · 2016-12-01T12:26:23Z

src/main/java/org/zalando/nakadi/service/ConsumerLimitingService.java

+                        && s.getEventType().equals(slot.getEventType()));
+        if (!hasMoreConnectionsToPartition) {
+            final String consumerPath = zkPathForConsumer(slot.getClient(), slot.getEventType(), slot.getPartition());
+            final PathChildrenCache cache = SLOTS_CACHES.getOrDefault(consumerPath, null);


You can do only remove since it returns removed value or null

Good catch! Fixed it here: 85ead49

adyach · 2016-12-01T13:48:38Z

…sumers

v-stepanov · 2016-12-12T10:50:24Z

@antban @rcillo @adyach @lmontrieux
This ticket is ready for a review again.
With latest changed I added a lock when creating a connection for client_id|event-type pair

rcillo · 2016-12-12T13:04:38Z

src/acceptance-test/java/org/zalando/nakadi/webservice/ConsumerLimitingServiceAT.java

+
+    @Test(expected = NoConnectionSlotsException.class)
+    public void whenNoFreeSlotsThenException() throws Exception {
+        final String parition = "0";


s/parition/partition/

fixed 9b5b4db

rcillo · 2016-12-12T13:23:10Z

src/main/java/org/zalando/nakadi/service/ConsumerLimitingCleaningService.java

+
+    private static final Charset CHARSET_UTF8 = Charset.forName("UTF-8");
+    private static final Logger LOG = LoggerFactory.getLogger(ConsumerLimitingCleaningService.class);
+    private static final int HANGING_NODES_CLEAN_PERIOD_MS = 6 * 60 * 60 * 1000; // 6 hours


What do you think of moving this number to the application config file?

I see. That's not possible because of the usage on annotation.

Yes, exactly. And even without that, I don't see much sense to make this property configurable.

rcillo · 2016-12-12T15:52:51Z

src/acceptance-test/java/org/zalando/nakadi/webservice/ConsumerLimitingCleaningServiceAT.java

+        deleteConsumersData();
+
+        // "hanging" node
+        CURATOR.create().creatingParentsIfNeeded().forPath("/nakadi/consumers/connections/hanging");


it wasn't clear for me that hanging is the name of a client. It took me some time to realise that it's the name of a client which has no consumers anymore. That's why you clean it. Changing it to consumer_with_no_open_streams could make if more explicit. This is just a suggestion to improve clarity.

Well, "hanging" here represents the whole node name that is usually combined from "client_id|event_type|partition". But in this test I test only ConsumerLimitingCleaningService class. And it is agnostic of node name, for this service the node is just a node. So I don't see much sense in this test to have a name close to real because ConsumerLimitingCleaningService doesn't care.

As I said, it's a suggestion to improve clarity. I took some time to understand it. I understand that for the machine it doesn't matter. But that's just me.

rcillo · 2016-12-12T16:18:18Z

rcillo · 2016-12-12T17:15:40Z

…sumers

rcillo · 2016-12-12T17:30:23Z

Vyacheslav Stepanov added 13 commits November 16, 2016 17:26

ARUHA-456: added 429 status code to consumption endpoint;

18a8bc5

ARUHA-456: added functionality to acquire/release/check the slots for…

b57cbd6

… consuming connections;

ARUHA-456: added functionality to acquire/release/check the slots for…

a7ad684

… consuming connections;

ARUHA-456: fixed compilation; fixed tests;

1415325

ARUHA-456: added acceptance test;

0685b52

ARUHA-456: added locking when getting connection slot in ZK;

bd9bfb7

ARUHA-456: added acceptance test for consumers limiting service;

1ab6979

ARUHA-456: fixed line size;

c22a863

ARUHA-456: fixed tests;

40c5b57

ARUHA-456: changed the way of acquiring slots in ZK;

402ecad

ARUHA-456: Added functionality to clean handing consumer nodes in ZK;

448cb22

ARUHA-456: Added tests for cleaning functionality;

1597418

ARUHA-456: removed using full class name;

41597a8

adyach reviewed Nov 22, 2016

View reviewed changes

Vyacheslav Stepanov added 2 commits November 22, 2016 16:36

ARUHA-456: fixed small issue in code;

37f2d31

ARUHA-456: fixed issue with static slot names list;

dedca7c

rcillo reviewed Nov 23, 2016

View reviewed changes

Vyacheslav Stepanov added 2 commits November 23, 2016 11:42

ARUHA-456: fixed loggers creation;

d82978a

ARUHA-456: extended test;

c8f0987

rcillo reviewed Nov 23, 2016

View reviewed changes

ARUHA-456: changed the number of retries in case of conflict;

fde52a9

antban reviewed Nov 23, 2016

View reviewed changes

Vyacheslav Stepanov added 6 commits November 30, 2016 15:02

ARUHA-456: fixed naming;

c6868d2

ARUHA-456: replaced hardcoded constaint in error message;

e8d7963

ARUHA-456: removed extra semicolon;

c9a4d5d

ARUHA-456: changed time constant;

c01f800

ARUHA-456: added some static imports;

c4825e2

ARUHA-456: fixed time diff calculation problem;

e903e15

Merge remote-tracking branch 'origin/master' into ARUHA-456-block-con…

6b24b47

…sumers

adyach reviewed Dec 1, 2016

View reviewed changes

ARUHA-456: fixed putting new cash to a map;

f41ca10

adyach reviewed Dec 1, 2016

View reviewed changes

ARUHA-456: improved code that closes cache;

85ead49

Vyacheslav Stepanov added 4 commits December 9, 2016 14:31

ARUHA-456: added locks;

5b0d416

ARUHA-456: fixed exception handling;

fd8fd94

ARUHA-456: minor fix;

af73756

Merge remote-tracking branch 'origin/master' into ARUHA-456-block-con…

cd34427

…sumers

rcillo reviewed Dec 12, 2016

View reviewed changes

ARUHA-456: fixed typo;

9b5b4db

Merge remote-tracking branch 'origin/master' into ARUHA-456-block-con…

cc44af0

…sumers

v-stepanov merged commit 2195cef into master Dec 13, 2016


		public static final String CONNECTIONS_ZK_PATH = "/nakadi/consumers/connections";

		private static final Logger LOG = LoggerFactory.getLogger(CursorsService.class);

Aruha 456 block consumers #489

Aruha 456 block consumers #489

Conversation

v-stepanov commented Nov 22, 2016 • edited

codecov-io commented Nov 22, 2016 • edited

Current coverage is 59.07% (diff: 14.41%)

adyach commented Nov 22, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

v-stepanov commented Nov 23, 2016

Choose a reason for hiding this comment

v-stepanov Nov 23, 2016 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

v-stepanov Nov 23, 2016 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rcillo commented Nov 23, 2016

antban Nov 23, 2016 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rcillo commented Nov 30, 2016

rcillo commented Nov 30, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

adyach Dec 1, 2016 • edited

Choose a reason for hiding this comment

v-stepanov Dec 1, 2016 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

adyach commented Dec 1, 2016

v-stepanov commented Dec 12, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rcillo commented Dec 12, 2016

rcillo commented Dec 12, 2016

rcillo commented Dec 12, 2016

v-stepanov commented Nov 22, 2016 •

edited

codecov-io commented Nov 22, 2016 •

edited

v-stepanov Nov 23, 2016 •

edited

v-stepanov Nov 23, 2016 •

edited

antban Nov 23, 2016 •

edited

adyach Dec 1, 2016 •

edited

v-stepanov Dec 1, 2016 •

edited