Aruha 473 check size of events #515

lmontrieux · 2016-12-16T09:55:51Z

This PR checks the size of all events in a batch. If there is at least one event that is over the maximum size after enrichment, then the entire batch is rejected with a 422.

v-stepanov · 2016-12-16T10:32:34Z

src/main/java/org/zalando/nakadi/domain/EventPublishingStep.java

@@ -4,6 +4,7 @@
    NONE,
    VALIDATING,
    ENRICHING,
+    VALIDATING_SIZE,


Unfortunately you can't change this. Because this enum is defined in our API and if you add values here that will mean that we add backwards incompatible changes.
https://github.com/zalando/nakadi/blob/master/api/nakadi-event-bus-api.yaml#L1817

Why do you want a separate step for that? I thought it can be part of VALIDATING step.

If we make it part of VALIDATING, then we will check the size of the event before enriching. I think it makes more sense to check the size after enriching, since the enrichment strategy is chose by the user who created the event type

I guess I could either check the size during validation, in which case we ignore whatever is added by the enrichment step; or, go back to validation after enriching, but that may be very confusing to users

I think the simplest way to go would be to assume that enrichment and metadata that is added by kafka will not add more data than the difference between nakadi limit (1kk bytes) and kafka limit (I think we wanted to set 2kk bytes)
In that case you can simply validate the size on validation step.

a061392 and 0a791da and df5b675

adyach · 2016-12-16T10:46:08Z

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

@@ -76,6 +83,9 @@ public EventPublishResult publish(final JSONArray events, final String eventType
        } catch (final EnrichmentException e) {
            LOG.debug("Event enrichment error: {}", e.getMessage());
            return aborted(EventPublishingStep.ENRICHING, batch);
+        } catch (final EventSizeValidationException e) {
+            LOG.debug("Event size validation error: ){", e.getMessage());


v-stepanov · 2016-12-16T10:46:58Z

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

+        for (int i = 0; i < 1000010; i++) {
+            sb.append("a");
+        }
+        sb.append("\"}]");


you can simply do: org.apache.commons.lang3.StringUtils.repeat("a", 1000010)

adyach · 2016-12-16T10:48:04Z

src/main/resources/application.yml

@@ -78,6 +78,7 @@ nakadi:
  auth:
    plugin:
      factory: org.zalando.nakadi.plugin.auth.DefaultApplicationServiceFactory
+  event.max.bytes: 1000000


I though we agreed on 1MB which is a bit bigger :)
1048576

I went for the default value in Kafka, which is 1,000,000, but I'm happy to go for a real 1MB if you think it is better

Also there is a problem with that, if you occasionally want to change producer's property message.max.bytes(and others related), let say to 3MB, it won't work because you restrict it to 1MB and you always have to remember about event.max.bytes.

https://kafka.apache.org/090/documentation.html#brokerconfigs
it is different according to docs :)
message.max.bytes= 1000012

Also there is a problem with that, if you occasionally want to change producer's property message.max.bytes(and others related), let say to 3MB, it won't work because you restrict it to 1MB and you always have to remember about event.max.bytes.

Why is it a problem?

because you increased kafka broker message size to 3MB and expect that it is gonna be 3MB, but actually it only 1MB

@adyach Why you expect that if you know that Nakadi validates event size as well?

today I expect that, because I remember :) it introduces another property to set when you try to change the size of the message and Kafka has a lot of them as well. if we could make it relative to Kafka producer it would be simpler to not forget.

@adyach I think the property in Nakadi should be primary and the property in Kafka should just satisfy the property in Nakadi. So if somebody will change in Kafka without looking to the value in Nakadi - he does something really wrong.

But I still don't really understand what the problem is.
One can say: what if somebody forgets how to operate AWS and will delete the whole kafka stack and we will loose all the data :) For me these are the problems of the same sort.

…ZE step

v-stepanov · 2016-12-16T12:31:51Z

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

            } catch (final EventValidationException e) {
                item.updateStatusAndDetail(EventPublishingStatus.FAILED, e.getMessage());
                throw e;
+            } catch (final EventSizeValidationException e) {
+                item.updateStatusAndDetail(EventPublishingStatus.ABORTED, e.getMessage());


I think here it should be FAILED like in a case when there was a schema validation problem. Because this is exactly the item that failed.

957938b and 49c1846

…r events in the batch are marked ABORTED

valgog · 2016-12-16T19:07:19Z

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

@@ -145,6 +151,11 @@ private void validateSchema(final JSONObject event, final EventType eventType) t
        }
    }

+    private void validateEventSize(final BatchItem item) throws EventValidationException {
+        if (item.getEvent().toString().getBytes().length > nakadiSettings.getEventMaxBytes())


Never use getBytes() without setting the default locale. Please always pass https://docs.oracle.com/javase/7/docs/api/java/nio/charset/StandardCharsets.html#UTF_8 object into that function.

Actually we have to check that we do not use cases where this stupid getByte() is done.

Another problem is to actually to do that conversion. There is the byte array representation that you can use. Or you should store the size into the item to avoid that conversion.

I'll fix this. There are a few other places in the code where we use getByte() without setting the locale, I'll create a ticket for that.

valgog · 2016-12-16T19:09:55Z

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

@@ -145,6 +151,11 @@ private void validateSchema(final JSONObject event, final EventType eventType) t
        }
    }

+    private void validateEventSize(final BatchItem item) throws EventValidationException {
+        if (item.getEvent().toString().getBytes().length > nakadiSettings.getEventMaxBytes())
+            throw new EventValidationException("Event too large");


Please provide a proper error message that includes the size and the limit.

@lmontrieux it seems that this message wasn't changed.

@rcillo Pretty sure I removed the whole method :). The size computation is now in the BatchItem constructor

@rcillo my mistake, I was wrong. Fixing it now

valgog · 2016-12-16T19:13:02Z

src/test/java/org/zalando/nakadi/service/EventPublisherTest.java

@@ -128,6 +139,147 @@ public void whenValidationFailsThenSubsequentItemsAreAborted() throws Exception
    }

    @Test
+    public void whenEventIsTooLargeThenResultIsAborted() throws Exception {


Please provide a test with multibyte chars in the message body.

There are two tests with multibyte characters already, lines 252-280. The first one is an event that is just 1 byte over the limit; the second one is exactly at the limit. Multibyte events are created on lines 502-512

valgog · 2016-12-18T11:04:00Z

@lmontrieux

After thinking about this more, it looks like we do not have an well defined idea how to limit the size of the messages.

The message itself has the size (in bytes).

The message is being enriched by information, that is also increasing the size of the message, written to Kafka.

This means that actual limitation on the customer message size is influenced by Kafka message limits and the enrichments, that we provide.

To be able to correctly show the reason for applying some limits on the customer message size, we have to also be able to extract and show information about added sizes by every provided enrichment. Such numbers would be also nice to have for monitoring anyway.

My suggestion currently would be:

extend Item object to keep both byte[] and String representation of the original message (check that conversions to and from String are done minimal amount of times)
extend Item object to keep statistics per enrichment (currently only added size in bytes per enrichment)
include information about the original and enriched sizes per enrichment into error message

What do you think?

v-stepanov · 2016-12-23T09:07:22Z

@valgog
I just wanted to mention that this was supposed to be a simple bugfix, not a spaceship-size feature.
Current situation is that message size is not checked anywhere and when somebody pushes message bigger than 1MB (kafka limit) - we respond with 207 (which is error situation for us) and have hole in the published batch.

The idea was very simple - to limit the message size to 1MB in Nakadi and 2MB in Kafka (so that 1MB + enrichment + kafka metadata will always fit to 2MB)

Also I have doubts about point 1 you suggested. That will in fact double Nakadi memory usage.

v-stepanov · 2016-12-23T09:07:48Z

sorry, I accidentally closed PR :)

lmontrieux · 2016-12-23T09:26:14Z

@valgog @v-stepanov I've had a short discussion offline with @valgog about this, but with Hack week I hadn't updated this PR yet. The way things work is:

producer sends an event of size x;
nakadi enriches the event, whose size is now x + e;
the kafka client serialises the event somehow, and add a bit of metadata. Size is now x + e + k.
kafka has 4 different configuration options for the max size of a message: the max size that a broker will accept (default 1MB); the max size of a log segment (must be higher than the max size of a message, default 1GB); the max size of data that a broker can replicate (default 1MB); the max size a consumer can read (default 1MB). Currently we use the defaults.

The idea behind this fix was to limit the size of an event to 1MB before enrichment (limit x only), but bump up all max values in Kafka to 2MB, to be safe (e+k are just a few bytes long). This way we will accept slightly larger messages than we do now. This only works assuming that increasing max values in Kafka will not increase its memory usage. I haven't looked into it but I'd be surprised in Kafka were to always allocate the max message size for each message.

As I recall, the outcome of this discussion was:
1/ limit size in validation, as originally planned;
2/ next week, as a different ticket, work on a more accurate way to limit the size of events, and keep statistics about the size of enrichments.

codecov-io · 2016-12-23T09:55:29Z

Current coverage is 59.57% (diff: 78.26%)

Merging #515 into master will increase coverage by 0.18%

@@             master       #515   diff @@
==========================================
  Files           198        198          
  Lines          5415       5470    +55   
  Methods           0          0          
  Messages          0          0          
  Branches        630        648    +18   
==========================================
+ Hits           3216       3259    +43   
- Misses         2017       2023     +6   
- Partials        182        188     +6

Powered by Codecov. Last update e03f6c8...e3a4562

lmontrieux · 2016-12-23T10:30:34Z

A quick look at the kafka code doesn't seem to indicate any allocation of max message size for each message.

…operty

rcillo · 2017-01-05T16:44:45Z

antban · 2017-01-06T08:02:24Z

api/nakadi-event-bus-api.yaml

@@ -316,7 +316,9 @@ paths:
        incoming Events. Validation rules are evaluated in the order they are defined and the Event
        is **rejected** in the first case of failure. If the offending validation rule provides
        information about the violation it will be included in the `BatchItemResponse`.  If the
-        `EventType` defines schema validation it will be performed at this moment.
+        `EventType` defines schema validation it will be performed at this moment. The size of each
+        Event will also be validated. The maximum size per Event is 1,000,000 bytes, before


@lmontrieux As far as I know, max.message.bytes for topic is 1,000,000 bytes. That means, that if we will enrich something, then it is guaranteed, that this message won't be written

@antban we had agreed to set it to something a bit lower, but I was waiting for today's standup to agree on a value. I've suggested 999,000 bytes on hipchat.

@antban 0330145

antban · 2017-01-06T08:16:20Z

src/main/java/org/zalando/nakadi/domain/BatchFactory.java

+        }
+
+        if (sb.length() != 0) {
+            batch.add(new BatchItem(sb.toString()));


Well, I completely disagree with this (I mean the fact that we are measuring user input), but there is a bug here:
It seems that message [ ] will create one batch item (but there is nothing there).

@antban You were right, there was a bug with [ ]. No batch would be created, but a JSONException would be thrown. I fixed it in 0330145

@antban As for measuring user input, I thought it was what we had agreed on earlier this week (or was it last week?). Happy to discuss it offline if I'm wrong.

rcillo · 2017-01-06T16:30:53Z

@lmontrieux is enrichment always using less than 100 bytes?

lmontrieux · 2017-01-06T16:36:11Z

@rcillo Well, we have (just under) 1,000 bytes of wiggle room for enrichment. Currently we only have one enrichment strategy, which sets: received_at, event_type, flow_id, partition, and version. Looks fairly stable in its size.

Of course, if we implement more enrichment strategies, we'll have to revise this.

rcillo · 2017-01-06T16:37:27Z

…, or spaces

rcillo

I remember you mentioning something about wrong status code. I could not find any code related to it.

rcillo · 2017-01-10T14:30:44Z

src/main/java/org/zalando/nakadi/domain/BatchFactory.java

+        final int length = events.length();
+        int end = length - 1;
+
+        while ((events.charAt(start) == ' '


I think this condition could be extracted to it's own method, something like isBlankCharacter.

or as per the test name isEmptyCharacter

Good point, will extract

rcillo · 2017-01-10T14:31:09Z

src/main/java/org/zalando/nakadi/domain/BatchFactory.java

+                && start < end) {
+            start++;
+        }
+        while ((events.charAt(end) == ' '


If this condition is extracted to its own method, then this should be refactored to reuse it.

lmontrieux · 2017-01-10T14:38:52Z

If a batch had "empty" (carriage return, tab, space,...) characters around it, it was rejected with a 400 for malformed data. When I made the comment I thought the issue was with large events, but investigating further, I found that it wasn't a size issue, but a trailing carriage return after the batch.

rcillo · 2017-01-10T15:15:12Z

rcillo · 2017-01-18T13:59:31Z

src/main/resources/application.yml

@@ -59,7 +59,6 @@ nakadi:
    send.timeoutMs: 5000
    batch.size: 5242880
    linger.ms: 0
-    fetch.message.max.bytes: 2000000


Why is this needed?

It is not needed anymore, since we decided to stick to the default value for Kafka. I forgot to remove it, and now I see in Scalyr that it was never correct in the first place, so this line is just ignored.

rcillo · 2017-01-18T15:57:26Z

lmontrieux · 2017-01-19T10:18:45Z

@rcillo I broke the build with my previous commit. This removes all code related to the update in the consumer's max fetch size (does not affect producer's max event size btw)

adyach · 2017-01-19T10:22:17Z

src/main/java/org/zalando/nakadi/repository/kafka/KafkaLocationManager.java

@@ -99,9 +99,7 @@ private void updateBrokers() {
    }

    public Properties getKafkaConsumerProperties() {
-        final Properties consumerProps = (Properties) kafkaProperties.clone();
-        consumerProps.put("fetch.message.max.bytes", kafkaSettings.getFetchMessageMaxBytes());


@lmontrieux I think we need to migrate to bigger message size

@adyach yes, but this is not working. I get this in the logs: [org.apache.kafka.clients.consumer.ConsumerConfig] --- The configuration fetch.message.max.bytes = 2000000 was supplied but isn't a known config.

I think we could do this as a separate ticket, where we increase the max event size from 999,000 bytes to 1,000,000 bytes

rcillo · 2017-01-19T12:41:44Z

v-stepanov · 2017-01-23T14:08:57Z

dismissed for not showing up in PR again :)

Lionel Montrieux added 5 commits December 15, 2016 15:05

ARUHA-473 Reject batches with at least one event that's too large

9dabfa2

ARUHA-473 Move maximum event size to nakadi settings

cc50d0c

ARUHA-473 Unit tests for max event size

761d455

ARUHA-473 Acceptance test for max event size

146d6a0

ARUHA-473 Fix checkstyle

8849782

v-stepanov reviewed Dec 16, 2016

View reviewed changes

adyach reviewed Dec 16, 2016

View reviewed changes

v-stepanov reviewed Dec 16, 2016

View reviewed changes

adyach reviewed Dec 16, 2016

View reviewed changes

Lionel Montrieux added 6 commits December 16, 2016 11:54

ARUHA-473 Move size validation after validation, remove VALIDATING_SI…

a061392

…ZE step

ARUHA-473 Move size validation inside validation

0a791da

ARUHA-473 Fix bug in test wrt event publishing step

df5b675

ARUHA-473 Refactor string building

e261cb8

ARUHA-473 Fix log string

217be7f

ARUHA-473 Set capacity for StringBuilder

ddd7a4f

v-stepanov reviewed Dec 16, 2016

View reviewed changes

Lionel Montrieux added 2 commits December 16, 2016 13:44

ARUHA-473 If an event is too large, that event is marked FAILED. Othe…

957938b

…r events in the batch are marked ABORTED

ARUHA-473 Remove redundant Exception

49c1846

valgog previously requested changes Dec 16, 2016

View reviewed changes

ARUHA-473 Explicit charset in getBytes()

cc15d1a

v-stepanov closed this Dec 23, 2016

v-stepanov reopened this Dec 23, 2016

ARUHA-473 Refactor event size computation

3f9bf87

ARUHA-473 Explicitely set Kafka consumers' fetch.message.max.bytes pr…

6f7a450

…operty

Lionel Montrieux added 3 commits January 5, 2017 17:54

ARUHA-473 Fix tests' expected error messages for events too large

f90e477

Merge branch 'master' into ARUHA-473-check-size-of-events

43a337a

ARUHA-473 Fix test

bf90d6e

antban reviewed Jan 6, 2017

View reviewed changes

Lionel Montrieux added 3 commits January 6, 2017 10:43

ARUHA-473 Change max event size in swagger file

0330145

ARUHA-473 Fix bug when a batch is an array with only spaces.

4cc0bcd

ARUHA-473 Set max event size to 999,000 bytes by default

30db7f5

ARUHA-473 Accept valid batches surrounded with tabs, carriage returns…

781db1e

…, or spaces

rcillo reviewed Jan 10, 2017

View reviewed changes

ARUHA-473 Refactoring

ffd729a

Lionel Montrieux added 2 commits January 18, 2017 12:07

Merge branch 'master' into ARUHA-473-check-size-of-events

e3a4562

ARUHA-473 Remove kafka max bytes property

58cb450

rcillo reviewed Jan 18, 2017

View reviewed changes

ARUHA-473 Finish removing kafka max bytes property

3f2a2ac

adyach reviewed Jan 19, 2017

View reviewed changes

Merge branch 'master' into ARUHA-473-check-size-of-events

57e67d3

lmontrieux merged commit 30c961b into master Jan 23, 2017

lmontrieux deleted the ARUHA-473-check-size-of-events branch January 24, 2017 09:57

Aruha 473 check size of events #515

Aruha 473 check size of events #515

Conversation

lmontrieux commented Dec 16, 2016

v-stepanov Dec 16, 2016 • edited Loading

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

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

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

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

valgog commented Dec 18, 2016

v-stepanov commented Dec 23, 2016

v-stepanov commented Dec 23, 2016

lmontrieux commented Dec 23, 2016

codecov-io commented Dec 23, 2016 • edited Loading

Current coverage is 59.57% (diff: 78.26%)

lmontrieux commented Dec 23, 2016

rcillo commented Jan 5, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

antban Jan 6, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rcillo commented Jan 6, 2017

lmontrieux commented Jan 6, 2017

rcillo commented Jan 6, 2017

rcillo left a 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

lmontrieux commented Jan 10, 2017

rcillo commented Jan 10, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rcillo commented Jan 18, 2017

lmontrieux commented Jan 19, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rcillo commented Jan 19, 2017

v-stepanov commented Jan 23, 2017

v-stepanov Dec 16, 2016 •

edited

Loading

codecov-io commented Dec 23, 2016 •

edited

Loading

antban Jan 6, 2017 •

edited

Loading