Added handling of NULL records and un-parseable records. #35
Conversation
…a and tests accordingly.
…d KafkaConsumerTest accordingly.
| @@ -254,16 +254,21 @@ public void accept(List<RecordOffset> recordOffsetObjectList) { | |||
| } | |||
|
|
|||
| byte[] byteArray = recordOffsetObject.getRecord(); // loads the byte[] contained in recordOffsetObject.getRecord() to byteArray. | |||
| batchBytes = batchBytes + byteArray.length; | |||
| if (byteArray == null) { | |||
There was a problem hiding this comment.
data error must cause a software to abort and abort in a way that it states why it aborted, invalid data must not be processed by default. please create a flag in configuration that allows invalid data to be skipped but logged.
There was a problem hiding this comment.
this flag should be regarding empty records.
There was a problem hiding this comment.
Added flag for empty records.
Solved in commit 9fb3949
| @@ -323,6 +328,13 @@ public void accept(List<RecordOffset> recordOffsetObjectList) { | |||
| catch (IOException e) { | |||
| throw new UncheckedIOException(e); | |||
| } | |||
| catch (ParseException e) { | |||
There was a problem hiding this comment.
invalid data must not be processed, please create a flag that allows parse failures to be skipped but logged.
There was a problem hiding this comment.
Added flag for non RFC5424 records.
Solved in commit 9fb3949
| // enable kerberus | ||
| conf.set("hadoop.security.authentication", config.getHadoopAuthentication()); | ||
| conf.set("hadoop.security.authorization", config.getHadoopAuthorization()); | ||
| conf.set("hadoop.kerberos.keytab.login.autorenewal.enabled", "true"); |
There was a problem hiding this comment.
make it configureable, please notice that some KDCs allow only limited amount of renewals and have a maximum total renewal time.
There was a problem hiding this comment.
Added configuration flag for autorenewal in commit 9d4311d
How should the ticket lifetime problem be approached? Can periodically using checkTGTAndReloginFromKeytab() on the UGI maintain the ticket indefinitely if automatic renewal won't do?
| @@ -97,9 +98,9 @@ public void readCoordinatorTest2Threads() { | |||
|
|
|||
| Thread.sleep(10000); | |||
| Assertions.assertEquals(2, messages.size()); | |||
| Assertions.assertEquals(140, messages.get(0).size() + messages.get(1).size()); // Assert that expected amount of records has been consumed by the consumer group. | |||
There was a problem hiding this comment.
2 new records were added to every test topic partition in MockKafkaConsumerFactory. One null record and one non rfc5424 record. With 10 topic partitions there are now 20 more records in the test data.
… Improved logging to give metadata for records causing the exceptions.
…les that weren't written to HDFS because of exception during processing.
| LOGGER.debug("Null record metadata: {}", recordOffsetObject.offsetToJSON()); | ||
| } | ||
| syslogFile.delete(); // Clean up | ||
| throw new NullPointerException("Record with null content detected during processing."); |
There was a problem hiding this comment.
throw custom exception instead of NPE because the case of null is handled by removing the file.
Added handling of NULL records and un-parseable records.
Partial fix for issue #30.