@Override
public void append(final LogEvent event) {
    try {
        final Layout<? extends Serializable> layout = getLayout();
        byte[] data;
        if (layout != null) {
            if (layout instanceof SerializedLayout) {
                final byte[] header = layout.getHeader();
                final byte[] body = layout.toByteArray(event);
                data = new byte[header.length + body.length];
                System.arraycopy(header, 0, data, 0, header.length);
                System.arraycopy(body, 0, data, header.length, body.length);
            } else {
                data = layout.toByteArray(event);
            }
        } else {
            data = StringEncoder.toBytes(event.getMessage().getFormattedMessage(), StandardCharsets.UTF_8);
        }
        manager.send(topic, data);
    } catch (final Exception e) {
        LOGGER.error("Unable to write to Kafka [{}] for appender [{}].", manager.getName(), getName(), e);
        throw new AppenderLoggingException("Unable to write to Kafka in appender: " + e.getMessage(), e);
    }
}
 

@Override
public void append(LogEvent event) {
    readLock.lock();
    try {
        final byte[] bytes = getLayout().toByteArray(event);
        registeredEvents.addFirst(new String(bytes));
        if (registeredEvents.size() > size) {
            registeredEvents.removeLast();
        }
    } catch (Exception ex) {
        if (!ignoreExceptions()) {
            throw new AppenderLoggingException(ex);
        }
    } finally {
        readLock.unlock();
    }
}
 

@Override
protected void writeInternal(final LogEvent event, final Serializable serializable) {
    if (!this.isRunning() || this.connection == null || this.connection.isClosed()) {
        throw new AppenderLoggingException(
                "Cannot write logging event; NoSQL manager not connected to the database.");
    }

    final NoSqlObject<W> entity = this.connection.createObject();
    if (serializable instanceof MapMessage) {
        setFields((MapMessage<?, ?>) serializable, entity);
    } else {
        setFields(event, entity);
    }

    this.connection.insertObject(entity);
}
 

@Override
protected synchronized void writeToDestination(final byte[] bytes, final int offset, final int length) {
    try {
        if (randomAccessFile == null) {
            String fileName = getFileName();
            File file = new File(fileName);
            FileUtils.makeParentDirs(file);
            createFileAfterRollover(fileName);
        }
        randomAccessFile.write(bytes, offset, length);
        size += length;
    } catch (final IOException ex) {
        final String msg = "Error writing to RandomAccessFile " + getName();
        throw new AppenderLoggingException(msg, ex);
    }
}
 

private boolean commitAndCloseAll() {
    if (this.connection != null || this.statement != null) {
        try {
            this.commitAndClose();
            return true;
        } catch (final AppenderLoggingException e) {
            // Database connection has likely gone stale.
            final Throwable cause = e.getCause();
            final Throwable actual = cause == null ? e : cause;
            logger().debug("{} committing and closing connection: {}", actual, actual.getClass().getSimpleName(),
                    e.toString(), e);
        }
    }
    if (factoryData.connectionSource != null) {
        factoryData.connectionSource.stop();
    }
    return true;
}
 

private void reconnectOn(final Exception exception) {
    if (!factoryData.retry) {
        throw new AppenderLoggingException("Cannot connect and prepare", exception);
    }
    if (reconnector == null) {
        reconnector = createReconnector();
        try {
            reconnector.reconnect();
        } catch (final SQLException reconnectEx) {
            logger().debug("Cannot reestablish JDBC connection to {}: {}; starting reconnector thread {}",
                    factoryData, reconnectEx, reconnector.getName(), reconnectEx);
            reconnector.start();
            reconnector.latch();
            if (connection == null || statement == null) {
                throw new AppenderLoggingException(exception, "Error sending to %s for %s [%s]", getName(),
                        factoryData, fieldsToString());
            }
        }
    }
}
 

/**
 * Sets the given Object in the prepared statement. The value is truncated if needed.
 */
private void setStatementObject(final int j, final String nameKey, final Object value) throws SQLException {
    if (statement == null) {
        throw new AppenderLoggingException("Cannot set a value when the PreparedStatement is null.");
    }
    if (value == null) {
        if (columnMetaData == null) {
            throw new AppenderLoggingException("Cannot set a value when the column metadata is null.");
        }
        // [LOG4J2-2762] [JDBC] MS-SQL Server JDBC driver throws SQLServerException when
        // inserting a null value for a VARBINARY column.
        // Calling setNull() instead of setObject() for null values fixes [LOG4J2-2762].
        this.statement.setNull(j, columnMetaData.get(nameKey).getType());
    } else {
        statement.setObject(j, truncate(nameKey, value));
    }
}
 

@Override
public void reportError(String message, Exception e) {

    if (e != null) {
        throw new AppenderLoggingException(e);
    }

    LOGGER.error(message, null, 0);
}
 

@Override
public void append(LogEvent event) {
    byte[] messageBytes = getLayout().toByteArray(event);
    MessageEntry messageEntry = new MessageEntry(messageBytes, event.getLevel());
    long nowTime = System.currentTimeMillis();
    List<MessageEntry> flushMessages;
    synchronized(bufferLock) { // TODO: REVIEW: maybe faster structure than synchronized to accumulate messages is possible...
        bufferMessages.add(messageEntry);
        bufferBytes += messageBytes.length;
        if ((bufferBytes < bufferMaxBytes) && ((nowTime - bufferLastFlushTime) < bufferMaxWait)) {
            return;
        }
        bufferBytes = 0;
        bufferLastFlushTime = nowTime;
        flushMessages = bufferMessages;
        bufferMessages = new ArrayList<>(bufferMinMsgCount);
    }
    // NOTE: Technically the below should be within synchronized, but it will slow down appends, and this alone is unlikely to result in too many ordering changes.
    try {
        List<Map<String, Object>> messageList = new ArrayList<>(flushMessages.size());
        for (MessageEntry entry : flushMessages) {
            messageList.add(entry.toMap());
        }
        Map<String, Object> dataMap = new HashMap<>();
        dataMap.put("messageList", messageList);
        JSON dataMapJson = JSON.from(dataMap);
        try {
            SocketSessionManager.allowLogging.set(false); // SPECIAL: Don't let this log anything or else endless logging
            SocketSessionManager.broadcastToChannel(dataMapJson.toString(), channel);
        } finally {
            SocketSessionManager.allowLogging.remove();
        }
    } catch (Exception ex) {
        if (!ignoreExceptions()) {
            throw new AppenderLoggingException(ex);
        }
    }
}
 

@Override
public void setStatus(Status status) {
  this.status = status;
  try {
    log.info(status.getMarker(), statusMessage);
  } catch (AppenderLoggingException e) {
    if (Main.isNotShuttingDown()) {
      log.error("Could not contact our internal Cassandra!!!" + e);
    }
  }
}
 

private void reportDocStatus(Status status, Document document, String message, Object... messageParams) {
  try {
    ThreadContext.put(JesterJAppender.JJ_INGEST_DOCID, document.getId());
    ThreadContext.put(JesterJAppender.JJ_INGEST_SOURCE_SCANNER, document.getSourceScannerName());
    document.setStatus(status);
    log.info(status.getMarker(), message, messageParams);
  } catch (AppenderLoggingException e) {
    if (Main.isNotShuttingDown()) {
      log.error("Could not contact our internal Cassandra!!!" + e);
    }
  } finally {
    ThreadContext.clearAll();
  }
}
 

@Override
public void insertObject(final NoSqlObject<Map<String, Object>> object) {
    try {
        client.index(object.unwrap());
    } catch (Exception e) {
        throw new AppenderLoggingException("failed to write log event to Elasticsearch: " + e.getMessage(), e);
    }
}
 

@Override
public void insertObject(final NoSqlObject<Map<String, Object>> object) {
    try {
        final Response response = this.client.save(object.unwrap());
        if (Strings.isNotEmpty(response.getError())) {
            throw new AppenderLoggingException(
                    "Failed to write log event to CouchDB due to error: " + response.getError() + '.');
        }
    } catch (final Exception e) {
        throw new AppenderLoggingException("Failed to write log event to CouchDB due to error: " + e.getMessage(),
                e);
    }
}
 

void send(final LogEvent event, final Serializable serializable) {
    if (messageProducer == null) {
        if (reconnector != null && !configuration.isImmediateFail()) {
            reconnector.latch();
            if (messageProducer == null) {
                throw new AppenderLoggingException(
                        "Error sending to JMS Manager '" + getName() + "': JMS message producer not available");
            }
        }
    }
    synchronized (this) {
        try {
            createMessageAndSend(event, serializable);
        } catch (final JMSException causeEx) {
            if (configuration.isRetry() && reconnector == null) {
                reconnector = createReconnector();
                try {
                    closeJndiManager();
                    reconnector.reconnect();
                } catch (NamingException | JMSException reconnEx) {
                    logger().debug("Cannot reestablish JMS connection to {}: {}; starting reconnector thread {}",
                            configuration, reconnEx.getLocalizedMessage(), reconnector.getName(), reconnEx);
                    reconnector.start();
                    throw new AppenderLoggingException(
                            String.format("JMS exception sending to %s for %s", getName(), configuration), causeEx);
                }
                try {
                    createMessageAndSend(event, serializable);
                } catch (final JMSException e) {
                    throw new AppenderLoggingException(
                            String.format("Error sending to %s after reestablishing JMS connection for %s", getName(),
                                    configuration),
                            causeEx);
                }
            }
        }
    }
}
 

@Override
public void insertObject(final NoSqlObject<Document> object) {
    try {
        final Document unwrapped = object.unwrap();
        LOGGER.debug("Inserting BSON Document {}", unwrapped);
        InsertOneResult insertOneResult = this.collection.insertOne(unwrapped);
        LOGGER.debug("Insert MongoDb result {}", insertOneResult);
    } catch (final MongoException e) {
        throw new AppenderLoggingException("Failed to write log event to MongoDB due to error: " + e.getMessage(),
                e);
    }
}
 

@Override
protected void connectAndStart() {
    try {
        this.entityManager = this.entityManagerFactory.createEntityManager();
        this.transaction = this.entityManager.getTransaction();
        this.transaction.begin();
    } catch (final Exception e) {
        throw new AppenderLoggingException(
                "Cannot write logging event or flush buffer; manager cannot create EntityManager or transaction.", e
        );
    }
}
 

@SuppressWarnings("sync-override") // synchronization on "this" is done within the method
@Override
protected void write(final byte[] bytes, final int offset, final int length, final boolean immediateFlush) {
    if (socket == null) {
        if (reconnector != null && !immediateFail) {
            reconnector.latch();
        }
        if (socket == null) {
            throw new AppenderLoggingException("Error writing to " + getName() + ": socket not available");
        }
    }
    synchronized (this) {
        try {
            writeAndFlush(bytes, offset, length, immediateFlush);
        } catch (final IOException causeEx) {
            if (retry && reconnector == null) {
                final String config = inetAddress + ":" + port;
                reconnector = createReconnector();
                try {
                    reconnector.reconnect();
                } catch (IOException reconnEx) {
                    LOGGER.debug("Cannot reestablish socket connection to {}: {}; starting reconnector thread {}",
                            config, reconnEx.getLocalizedMessage(), reconnector.getName(), reconnEx);
                    reconnector.start();
                    throw new AppenderLoggingException(
                            String.format("Error sending to %s for %s", getName(), config), causeEx);
                }
                try {
                    writeAndFlush(bytes, offset, length, immediateFlush);
                } catch (IOException e) {
                    throw new AppenderLoggingException(
                            String.format("Error writing to %s after reestablishing connection for %s", getName(),
                                    config),
                            causeEx);
                }
            }
        }
    }
}
 

@Override
protected void connectAndStart() {
    try {
        this.connection = this.provider.getConnection();
    } catch (final Exception e) {
        throw new AppenderLoggingException("Failed to get connection from NoSQL connection provider.", e);
    }
}
 

@Test
public void testWriteInternalNotConnected01() {
    try (final NoSqlDatabaseManager<?> manager = NoSqlDatabaseManager.getNoSqlDatabaseManager("name", 0,
        provider)) {
        expectedException.expect(AppenderLoggingException.class);
        manager.writeInternal(mock(LogEvent.class), null);
    }
}
 

@Test
public void testWriteInternalNotConnected02() {
    given(connection.isClosed()).willReturn(true);

    try (final NoSqlDatabaseManager<?> manager = NoSqlDatabaseManager.getNoSqlDatabaseManager("name", 0,
        provider)) {

        manager.startup();
        manager.connectAndStart();
        then(provider).should().getConnection();

        expectedException.expect(AppenderLoggingException.class);
        manager.writeInternal(mock(LogEvent.class), null);
    }
}
 

@Override
public void insertObject(final NoSqlObject<Document> object) {
    try {
        final Document unwrapped = object.unwrap();
        LOGGER.debug("Inserting object {}", unwrapped);
        this.collection.insertOne(unwrapped);
    } catch (final MongoException e) {
        throw new AppenderLoggingException("Failed to write log event to MongoDB due to error: " + e.getMessage(),
                e);
    }
}
 

@Override
public void append(LogEvent event) {
    final Layout<? extends Serializable> layout = getLayout();
    final String formattedMessage;
    if (layout == null) {
        formattedMessage = event.getMessage().getFormattedMessage();
    } else {
        formattedMessage = new String(layout.toByteArray(event), StandardCharsets.UTF_8);
    }

    final GelfMessageBuilder builder = new GelfMessageBuilder(formattedMessage, hostName)
            .timestamp(event.getTimeMillis() / 1000d)
            .level(GelfMessageLevel.fromNumericLevel(Severity.getSeverity(event.getLevel()).getCode()))
            .additionalField("loggerName", event.getLoggerName())
            .additionalField("threadName", event.getThreadName());

    final Marker marker = event.getMarker();
    if (marker != null) {
        builder.additionalField("marker", marker.getName());
    }

    if (includeThreadContext) {
        for (Map.Entry<String, String> entry : event.getContextMap().entrySet()) {
            builder.additionalField(entry.getKey(), entry.getValue());
        }

        // Guard against https://issues.apache.org/jira/browse/LOG4J2-1530
        final ThreadContext.ContextStack contextStack = event.getContextStack();
        if (contextStack != null) {
            final List<String> contextStackItems = contextStack.asList();
            if (contextStackItems != null && !contextStackItems.isEmpty()) {
                builder.additionalField("contextStack", contextStackItems.toString());
            }
        }
    }

    if (includeSource) {
        final StackTraceElement source = event.getSource();
        if (source != null) {
            builder.additionalField("sourceFileName", source.getFileName());
            builder.additionalField("sourceMethodName", source.getMethodName());
            builder.additionalField("sourceClassName", source.getClassName());
            builder.additionalField("sourceLineNumber", source.getLineNumber());
        }
    }

    @SuppressWarnings("all")
    final Throwable thrown = event.getThrown();
    if (includeStackTrace && thrown != null) {
        String stackTrace;
        if (includeExceptionCause) {
            final StringWriter stringWriter = new StringWriter();
            final PrintWriter printWriter = new PrintWriter(stringWriter);
            thrown.printStackTrace(printWriter);
            stackTrace = stringWriter.toString();
        } else {
            stackTrace = getSimpleStacktraceAsString(thrown);
        }

        builder.additionalField("exceptionClass", thrown.getClass().getCanonicalName());
        builder.additionalField("exceptionMessage", thrown.getMessage());
        builder.additionalField("exceptionStackTrace", stackTrace);

        builder.fullMessage(formattedMessage);
    }

    if (!additionalFields.isEmpty()) {
        builder.additionalFields(additionalFields);
    }

    final GelfMessage gelfMessage = builder.build();
    try {
        final boolean sent = client.trySend(gelfMessage);
        if (!sent) {
            LOG.debug("Couldn't send message: {}", gelfMessage);
        }
    } catch (Exception e) {
        throw new AppenderLoggingException("failed to write log event to GELF server: " + e.getMessage(), e);
    }
}
 

private void handleError(final String prefix) {
    final String msg = appenderErrorHandlerMessage(prefix);
    if (!appender.ignoreExceptions()) {
        throw new AppenderLoggingException(msg);
    }
}
 

@Test
public void testReconnect() throws Exception {
    list.clear();
    resolver.ports = new int[] {ports[0]};
    server1 = new TestSocketServer(ports[0]);
    server1.start();
    Thread.sleep(200);
    String message = "Log #1";
    String msg = null;
    for (int i = 0; i < 5; ++i) {
        logger.error(message);
        Thread.sleep(100);
        if (list.size() > 0) {
            msg = list.get(0);
            if (msg != null) {
                break;
            }
        }
    }
    assertNotNull("No message", msg);
    assertEquals(message, msg);

    logger.error(SHUTDOWN);
    server1.join();

    list.clear();

    message = "Log #2";
    boolean exceptionCaught = false;

    for (int i = 0; i < 100; ++i) {
        try {
            logger.error(message);
        } catch (final AppenderLoggingException e) {
            exceptionCaught = true;
            break;
            // System.err.println("Caught expected exception");
        }
    }
    assertTrue("No Exception thrown", exceptionCaught);
    message = "Log #3";


    server1 = new TestSocketServer(ports[0]);
    server1.start();
    Thread.sleep(300);

    msg = null;
    for (int i = 0; i < 5; ++i) {
        logger.error(message);
        Thread.sleep(100);
        if (list.size() > 0) {
            msg = list.get(0);
            if (msg != null) {
                break;
            }
        }
    }
    assertNotNull("No message", msg);
    assertEquals(message, msg);
    logger.error(SHUTDOWN);
    server1.join();
}