@Test public void testWriterWithHeadersInjectorEnabledWithWrongClass() throws Exception { SingerLog singerLog = new SingerLog(createSingerLogConfig("test", "/a/b/c")); LogStream logStream = new LogStream(singerLog, "test.tmp"); KafkaMessagePartitioner partitioner = new Crc32ByteArrayPartitioner(); KafkaProducerConfig config = new KafkaProducerConfig(); SingerSettings.setSingerConfig(new SingerConfig()); KafkaProducerManager.injectTestProducer(config, producer); logStream.getSingerLog().getSingerLogConfig().setHeadersInjectorClass("com.pinterest.x.y.z"); @SuppressWarnings("resource") KafkaWriter writer = new KafkaWriter(logStream, config, partitioner, "topicx", false, Executors.newCachedThreadPool(), true); assertNull(writer.getHeadersInjector()); }
@VisibleForTesting public HeadersInjector getHeadersInjector() { return headersInjector; }
KafkaWriter implements LogStreamWriter { @VisibleForTesting public HeadersInjector getHeadersInjector() { return headersInjector; } }
KafkaWriter implements LogStreamWriter { @VisibleForTesting public HeadersInjector getHeadersInjector() { return headersInjector; } KafkaWriter(LogStream logStream,
                     KafkaProducerConfig producerConfig,
                     String topic,
                     boolean skipNoLeaderPartitions,
                     boolean auditingEnabled,
                     String auditTopic,
                     String partitionerClassName,
                     int writeTimeoutInSeconds);  KafkaWriter(LogStream logStream,
                     KafkaProducerConfig producerConfig,
                     String topic,
                     boolean skipNoLeaderPartitions,
                     boolean auditingEnabled,
                     String auditTopic,
                     String partitionerClassName,
                     int writeTimeoutInSeconds, boolean enableHeadersInjector); @VisibleForTesting protected  KafkaWriter(LogStream logStream,
                        KafkaProducerConfig producerConfig,
                        KafkaMessagePartitioner partitioner,
                        String topic,
                        boolean skipNoLeaderPartitions,
                        ExecutorService clusterThreadPool,
                        boolean enableHeadersInjector); protected  KafkaWriter(KafkaProducerConfig producerConfig,
                        KafkaMessagePartitioner partitioner,
                        String topic,
                        boolean skipNoLeaderPartitions,
                        ExecutorService clusterThreadPool); }
KafkaWriter implements LogStreamWriter { @VisibleForTesting public HeadersInjector getHeadersInjector() { return headersInjector; } KafkaWriter(LogStream logStream,
                     KafkaProducerConfig producerConfig,
                     String topic,
                     boolean skipNoLeaderPartitions,
                     boolean auditingEnabled,
                     String auditTopic,
                     String partitionerClassName,
                     int writeTimeoutInSeconds);  KafkaWriter(LogStream logStream,
                     KafkaProducerConfig producerConfig,
                     String topic,
                     boolean skipNoLeaderPartitions,
                     boolean auditingEnabled,
                     String auditTopic,
                     String partitionerClassName,
                     int writeTimeoutInSeconds, boolean enableHeadersInjector); @VisibleForTesting protected  KafkaWriter(LogStream logStream,
                        KafkaProducerConfig producerConfig,
                        KafkaMessagePartitioner partitioner,
                        String topic,
                        boolean skipNoLeaderPartitions,
                        ExecutorService clusterThreadPool,
                        boolean enableHeadersInjector); protected  KafkaWriter(KafkaProducerConfig producerConfig,
                        KafkaMessagePartitioner partitioner,
                        String topic,
                        boolean skipNoLeaderPartitions,
                        ExecutorService clusterThreadPool); boolean isEnableLoggingAudit(); AuditHeadersGenerator getAuditHeadersGenerator(); @VisibleForTesting void setAuditHeadersGenerator(
      AuditHeadersGenerator auditHeadersGenerator); AuditConfig getAuditConfig(); @Override LogStream getLogStream(); @Override boolean isAuditingEnabled(); @VisibleForTesting HeadersInjector getHeadersInjector(); @VisibleForTesting ExecutorService getClusterThreadPool(); boolean checkMessageValidAndInjectHeaders(
          LogMessage msg, Headers headers, int indexWithinTheBucket, int partitionId,
          Map<Integer, Map<Integer, LoggingAuditHeaders>> mapOfHeadersMaps,
          Map<Integer, Map<Integer, Boolean>> mapOfMessageValidMaps); void checkAndSetLoggingAuditHeadersForLogMessage(LogMessage msg); @Override void writeLogMessages(List<LogMessage> logMessages); void enqueueLoggingAuditEvents(KafkaWritingTaskResult result,
                                        Map<Integer, LoggingAuditHeaders> headersMap,
                                        Map<Integer, Boolean> messageValidMap); void close(); }
KafkaWriter implements LogStreamWriter { @VisibleForTesting public HeadersInjector getHeadersInjector() { return headersInjector; } KafkaWriter(LogStream logStream,
                     KafkaProducerConfig producerConfig,
                     String topic,
                     boolean skipNoLeaderPartitions,
                     boolean auditingEnabled,
                     String auditTopic,
                     String partitionerClassName,
                     int writeTimeoutInSeconds);  KafkaWriter(LogStream logStream,
                     KafkaProducerConfig producerConfig,
                     String topic,
                     boolean skipNoLeaderPartitions,
                     boolean auditingEnabled,
                     String auditTopic,
                     String partitionerClassName,
                     int writeTimeoutInSeconds, boolean enableHeadersInjector); @VisibleForTesting protected  KafkaWriter(LogStream logStream,
                        KafkaProducerConfig producerConfig,
                        KafkaMessagePartitioner partitioner,
                        String topic,
                        boolean skipNoLeaderPartitions,
                        ExecutorService clusterThreadPool,
                        boolean enableHeadersInjector); protected  KafkaWriter(KafkaProducerConfig producerConfig,
                        KafkaMessagePartitioner partitioner,
                        String topic,
                        boolean skipNoLeaderPartitions,
                        ExecutorService clusterThreadPool); boolean isEnableLoggingAudit(); AuditHeadersGenerator getAuditHeadersGenerator(); @VisibleForTesting void setAuditHeadersGenerator(
      AuditHeadersGenerator auditHeadersGenerator); AuditConfig getAuditConfig(); @Override LogStream getLogStream(); @Override boolean isAuditingEnabled(); @VisibleForTesting HeadersInjector getHeadersInjector(); @VisibleForTesting ExecutorService getClusterThreadPool(); boolean checkMessageValidAndInjectHeaders(
          LogMessage msg, Headers headers, int indexWithinTheBucket, int partitionId,
          Map<Integer, Map<Integer, LoggingAuditHeaders>> mapOfHeadersMaps,
          Map<Integer, Map<Integer, Boolean>> mapOfMessageValidMaps); void checkAndSetLoggingAuditHeadersForLogMessage(LogMessage msg); @Override void writeLogMessages(List<LogMessage> logMessages); void enqueueLoggingAuditEvents(KafkaWritingTaskResult result,
                                        Map<Integer, LoggingAuditHeaders> headersMap,
                                        Map<Integer, Boolean> messageValidMap); void close(); static final String HOSTNAME; }