StreamAppender extends AppenderSkeleton { @Override public void activateOptions() { String containerName = System.getProperty(JAVA_OPTS_CONTAINER_NAME); if (containerName != null) { isApplicationMaster = containerName.contains(JOB_COORDINATOR_TAG); } else { throw new SamzaException("Got null container name from system property: " + JAVA_OPTS_CONTAINER_NAME + ". This is used as the key for the log appender, so can't proceed."); } key = containerName; if (isApplicationMaster) { systemInitialized = false; } else { setupSystem(); systemInitialized = true; } }  String getStreamName(); void setStreamName(String streamName); int getPartitionCount(); void setPartitionCount(int partitionCount); @Override void activateOptions(); @Override void append(LoggingEvent event); @Override void close(); @Override boolean requiresLayout(); void flushSystemProducer(); Serde<LoggingEvent> getSerde();  }
@Test public void testSystemProducerAppenderInContainer() throws InterruptedException { System.setProperty("samza.container.name", "samza-container-1"); MockSystemProducerAppender systemProducerAppender = new MockSystemProducerAppender(); PatternLayout layout = new PatternLayout(); layout.setConversionPattern("%m"); systemProducerAppender.setLayout(layout); systemProducerAppender.activateOptions(); log.addAppender(systemProducerAppender); List<String> messages = Lists.newArrayList("testing1", "testing2"); logAndVerifyMessages(messages); }
@Test public void testNoStreamCreationUponSetupByDefault() { System.setProperty("samza.container.name", "samza-container-1"); MockSystemProducerAppender systemProducerAppender = new MockSystemProducerAppender(); PatternLayout layout = new PatternLayout(); layout.setConversionPattern("%m"); systemProducerAppender.setLayout(layout); systemProducerAppender.activateOptions(); log.addAppender(systemProducerAppender); Assert.assertEquals("", MockSystemAdmin.createdStreamName); }
@Test public void testStreamCreationUpSetupWhenEnabled() { System.setProperty("samza.container.name", "samza-container-1"); MapConfig mapConfig = new MapConfig(ImmutableMap.of( "task.log4j.create.stream.enabled", "true", "job.name", "log4jTest", "job.id", "1", "systems.mock.samza.factory", MockSystemFactory.class.getCanonicalName(), "task.log4j.system", "mock")); MockSystemProducerAppender systemProducerAppender = new MockSystemProducerAppender(mapConfig); PatternLayout layout = new PatternLayout(); layout.setConversionPattern("%m"); systemProducerAppender.setLayout(layout); systemProducerAppender.activateOptions(); log.addAppender(systemProducerAppender); Assert.assertEquals("__samza_log4jTest_1_logs", MockSystemAdmin.createdStreamName); }
@Test public void testExceptionsDoNotKillTransferThread() throws InterruptedException { System.setProperty("samza.container.name", "samza-container-1"); MockSystemProducerAppender systemProducerAppender = new MockSystemProducerAppender(); PatternLayout layout = new PatternLayout(); layout.setConversionPattern("%m"); systemProducerAppender.setLayout(layout); systemProducerAppender.activateOptions(); log.addAppender(systemProducerAppender); List<String> messages = Lists.newArrayList("testing5", "testing6", "testing7"); final CountDownLatch allMessagesSent = new CountDownLatch(messages.size()); MockSystemProducer.listeners.add((source, envelope) -> { allMessagesSent.countDown(); if (allMessagesSent.getCount() == messages.size() - 1) { throw new RuntimeException(); } }); messages.forEach((message) -> log.info(message)); assertTrue("Thread did not send all messages. Count: " + allMessagesSent.getCount(), allMessagesSent.await(60, TimeUnit.SECONDS)); }
@Test public void testQueueTimeout() throws InterruptedException { System.setProperty("samza.container.name", "samza-container-1"); MockSystemProducerAppender systemProducerAppender = new MockSystemProducerAppender(); systemProducerAppender.queueTimeoutS = 1; PatternLayout layout = new PatternLayout(); layout.setConversionPattern("%m"); systemProducerAppender.setLayout(layout); systemProducerAppender.activateOptions(); log.addAppender(systemProducerAppender); int extraMessageCount = 5; int expectedMessagesSent = extraMessageCount - 1; List<String> messages = new ArrayList<>(StreamAppender.DEFAULT_QUEUE_SIZE + extraMessageCount); for (int i = 0; i < StreamAppender.DEFAULT_QUEUE_SIZE + extraMessageCount; i++) { messages.add(String.valueOf(i)); } final CountDownLatch allMessagesSent = new CountDownLatch(expectedMessagesSent); final CountDownLatch waitForTimeout = new CountDownLatch(1); MockSystemProducer.listeners.add((source, envelope) -> { allMessagesSent.countDown(); try { waitForTimeout.await(); } catch (InterruptedException e) { fail("Test could not run properly because of a thread interrupt."); } }); messages.forEach((message) -> log.info(message)); assertEquals(messages.size() - expectedMessagesSent, systemProducerAppender.metrics.logMessagesDropped.getCount()); waitForTimeout.countDown(); assertTrue("Thread did not send all messages. Count: " + allMessagesSent.getCount(), allMessagesSent.await(60, TimeUnit.SECONDS)); assertEquals(expectedMessagesSent, MockSystemProducer.messagesReceived.size()); }