[samza] branch master updated: Make StreamAppender extensible for sending event to SystemProducer (#1627)

[bharathkk]

This is an automated email from the ASF dual-hosted git repository.

bharathkk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/samza.git


The following commit(s) were added to refs/heads/master by this push:
     new 4e4e477bf Make StreamAppender extensible for sending event to 
SystemProducer (#1627)
4e4e477bf is described below

commit 4e4e477bf14d86a9784399d44fd7ae70b7705b69
Author: jia-gao <94939653+jia-...@users.noreply.github.com>
AuthorDate: Mon Sep 19 10:25:28 2022 -0700

    Make StreamAppender extensible for sending event to SystemProducer (#1627)
    
    Issues:
    Currently, StreamAppender is not extensible in terms of setting 
systemProducer and exposing sendEventToSystemProducer to child classes. It also 
makes it hard for child classes out of package to unit test any logic related 
to systemProducer
    
    Changes: Make StreamAppender extensible
    1. Change method sendEventToSystemProducer from private to protected
    2. Change member variable systemProducer from private to protected
---
 .../main/java/org/apache/samza/logging/log4j2/StreamAppender.java   | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git 
a/samza-log4j2/src/main/java/org/apache/samza/logging/log4j2/StreamAppender.java
 
b/samza-log4j2/src/main/java/org/apache/samza/logging/log4j2/StreamAppender.java
index aab712af8..c369282b8 100644
--- 
a/samza-log4j2/src/main/java/org/apache/samza/logging/log4j2/StreamAppender.java
+++ 
b/samza-log4j2/src/main/java/org/apache/samza/logging/log4j2/StreamAppender.java
@@ -73,7 +73,7 @@ public class StreamAppender extends AbstractAppender {
   private final BlockingQueue<EncodedLogEvent> logQueue = new 
LinkedBlockingQueue<>(DEFAULT_QUEUE_SIZE);
 
   private SystemStream systemStream = null;
-  private SystemProducer systemProducer = null;
+  protected SystemProducer systemProducer = null;
   private String key = null;
   private byte[] keyBytes; // Serialize the key once, since we will use it for 
every event.
   private String containerName = null;
@@ -416,7 +416,7 @@ public class StreamAppender extends AbstractAppender {
    * Helper method to send a serialized log-event to the systemProducer, and 
increment respective methods.
    * @param logQueueEntry the serialized log-event to be sent to the 
systemProducer
    */
-  private void sendEventToSystemProducer(EncodedLogEvent logQueueEntry) {
+  protected void sendEventToSystemProducer(EncodedLogEvent logQueueEntry) {
     metrics.logMessagesBytesSent.inc(logQueueEntry.getEntryValueSize());
     metrics.logMessagesCountSent.inc();
     systemProducer.send(SOURCE, decorateLogEvent(logQueueEntry));
@@ -513,4 +513,4 @@ public class StreamAppender extends AbstractAppender {
       return entryValue.length;
     }
   }
-}
\ No newline at end of file
+}