Do not fail on write error

[pdambrauskas]

Fixes #1505

This issue occurred to us too. Found existing issue about it and made this PR.

• Moved offset parsing and message parsing to single try/catch block
• Reused fail backoff logic which was implemented in parse block
• Made fail threshold configurable
• Split consumeNextMessage to multiple smaller methods to make it more readable.

[HenryCaiHaiying]

The refactoring makes the code diff difficult to read, I wasn't sure the purpose of making those new methods. Looks like the main purpose of the PR is to add mBasMessages, you can just insert this logic in place. Also what kind of errors do you see during messageWriter.write and you want to skip those errors?

[HenryCaiHaiying]

Did we lose the reset of DECAY after refactoring?

[pdambrauskas]

DECAY is final and used to decrease mBadMessages as before. It does not make sense to declare new DECAY variable on each writeMessage call, so I moved it to static final constant.

[pdambrauskas]

The refactoring makes the code diff difficult to read, I wasn't sure the purpose of making those new methods. Looks like the main purpose of the PR is to add mBasMessages, you can just insert this logic in place. Also what kind of errors do you see during messageWriter.write and you want to skip those errors?

Hey Henry, thanks for taking time to review my Pull Requests :). I do understand that it is hard to review this code, since the method was not that small.

mBadMessages was introduced to make it possible to ignore all failing messages, but it is not the main purpose of this PR.

To simplify the idea, before code looked like this:

//...
try {
  // parse partition
} catch (Exception e) {
  // increment consumer.message_errors.count metric
  // handle fail, re-throw if there were > 1000 errors 
}

try {
  // write message
} catch (Exception e) {
  // just re-throw
}

After this change there is only one try block:

try {
  // parse partition
  // write message
} catch (Exception e) {
  // increment consumer.message_errors.count metric
  // handle fail, re-throw if there were > 1000 errors 
}

This allows us to configure secor to not fail on writes and also onsumer.message_errors.count is more correct.

Also what kind of errors do you see during messageWriter.write and you want to skip those errors?

Our error is very similar to one, described in #1505. We use custom ReaderWriterFactory implementation with custom Writer to write json messages to parquet and validate schema, so exception is a bit different though, I thought it would make sense to catch exception in our Writer, but then I saw that others have the same problem on #1505. And consumer.message_errors.count metric is increased on Consumer class, so i decided to keep on place responsible for handling this kind of errors.

[HenryCaiHaiying]

Your explanation makes sense, I will merge in your PR, thanks.