test_class.declarator: processor
test_class.modifier: private
test_class.original_string: private KafkaWriter processor;
test_class.type: KafkaWriter
test_class.var_name: processor
test_class.declarator: producer
test_class.modifier: private
test_class.original_string: private kafka.javaapi.producer.Producer<String, String> producer;
test_class.type: kafka.javaapi.producer.Producer<String, String>
test_class.var_name: producer
test_class.file: fabric-components/kafka-writer/src/test/java/com/olacabs/fabric/processors/kafkawriter/KafkaWriterTest.java
test_class.identifier: KafkaWriterTest
test_case.body: @Test
      public void testConsume() throws Exception {
          Random rnd = new Random();
          ObjectMapper mapper = new ObjectMapper();
          List<String> lst = Lists.newArrayList();
          for (int i = 0; i < 5; i++) {
              lst.add("{\"firstName\":\"" + "xyz" + rnd.nextInt(10) + "\", \"lastName\":\"Doe" + rnd.nextInt(100)
                      + "\"}");
  
          }
  
          ProcessorTestBench processorTestBench = new ProcessorTestBench(false);
          List<EventSet> events = processorTestBench.runStreamingProcessor(processor, ImmutableList
                  .of(EventSet.eventFromEventBuilder().events(ImmutableList
                                  .of(Event.builder().jsonNode(mapper.readTree(lst.get(0))).build(),
                                          Event.builder().jsonNode(mapper.readTree(lst.get(0))).build())).build(),
                          EventSet.eventFromEventBuilder().events(ImmutableList
                                  .of(Event.builder().jsonNode(mapper.readTree(lst.get(0))).build(),
                                          Event.builder().jsonNode(mapper.readTree(lst.get(0))).build(),
                                          Event.builder().jsonNode(mapper.readTree(lst.get(0))).build())).build()));
  
          verify(producer, times(2)).send(anyList());
  
  
      }
test_case.class_method_signature: KafkaWriterTest.testConsume()
test_case.full_signature: @Test public void testConsume()
test_case.identifier: testConsume
test_case.invocations: newArrayList add nextInt nextInt runStreamingProcessor of build events eventFromEventBuilder of build jsonNode builder readTree get build jsonNode builder readTree get build events eventFromEventBuilder of build jsonNode builder readTree get build jsonNode builder readTree get build jsonNode builder readTree get send verify times anyList
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testConsume()
focal_class.declarator: LOGGER = LoggerFactory.getLogger(KafkaWriter.class.getSimpleName())
focal_class.modifier: private static final
focal_class.original_string: private static final Logger LOGGER = LoggerFactory.getLogger(KafkaWriter.class.getSimpleName());
focal_class.type: Logger
focal_class.var_name: LOGGER
focal_class.declarator: DEFAULT_IGNORE_SERIALIZATION_ERROR = false
focal_class.modifier: private static final
focal_class.original_string: private static final boolean DEFAULT_IGNORE_SERIALIZATION_ERROR = false;
focal_class.type: boolean
focal_class.var_name: DEFAULT_IGNORE_SERIALIZATION_ERROR
focal_class.declarator: DEFAULT_TOPIC_ON_JSON_PATH = false
focal_class.modifier: private static final
focal_class.original_string: private static final boolean DEFAULT_TOPIC_ON_JSON_PATH = false;
focal_class.type: boolean
focal_class.var_name: DEFAULT_TOPIC_ON_JSON_PATH
focal_class.declarator: DEFAULT_SERIALIZER_CLASS = "kafka.serializer.StringEncoder"
focal_class.modifier: private static final
focal_class.original_string: private static final String DEFAULT_SERIALIZER_CLASS = "kafka.serializer.StringEncoder";
focal_class.type: String
focal_class.var_name: DEFAULT_SERIALIZER_CLASS
focal_class.declarator: DEFAULT_KAFKA_KEY_JSON_PATH = "/metadata/partitionKey/value"
focal_class.modifier: private static final
focal_class.original_string: private static final String DEFAULT_KAFKA_KEY_JSON_PATH = "/metadata/partitionKey/value";
focal_class.type: String
focal_class.var_name: DEFAULT_KAFKA_KEY_JSON_PATH
focal_class.declarator: DEFAULT_ACK_COUNT = 1
focal_class.modifier: private static final
focal_class.original_string: private static final int DEFAULT_ACK_COUNT = 1;
focal_class.type: int
focal_class.var_name: DEFAULT_ACK_COUNT
focal_class.declarator: DEFAULT_BATCH_SIZE = 10
focal_class.modifier: private static final
focal_class.original_string: private static final int DEFAULT_BATCH_SIZE = 10;
focal_class.type: int
focal_class.var_name: DEFAULT_BATCH_SIZE
focal_class.declarator: ACK_COUNT = "-1"
focal_class.modifier: private static final
focal_class.original_string: private static final String ACK_COUNT = "-1";
focal_class.type: String
focal_class.var_name: ACK_COUNT
focal_class.declarator: kafkaKeyJsonPath
focal_class.modifier: private
focal_class.original_string: private String kafkaKeyJsonPath;
focal_class.type: String
focal_class.var_name: kafkaKeyJsonPath
focal_class.declarator: ignoreError
focal_class.modifier: private
focal_class.original_string: private boolean ignoreError;
focal_class.type: boolean
focal_class.var_name: ignoreError
focal_class.declarator: mapper
focal_class.modifier: private
focal_class.original_string: private ObjectMapper mapper;
focal_class.type: ObjectMapper
focal_class.var_name: mapper
focal_class.declarator: kafkaTopic
focal_class.modifier: @Getter
      @Setter
      private
focal_class.original_string: @Getter
      @Setter
      private String kafkaTopic;
focal_class.type: String
focal_class.var_name: kafkaTopic
focal_class.declarator: kafkaTopicJsonPath
focal_class.modifier: @Getter
      @Setter
      private
focal_class.original_string: @Getter
      @Setter
      private String kafkaTopicJsonPath;
focal_class.type: String
focal_class.var_name: kafkaTopicJsonPath
focal_class.declarator: ingestionPoolSize
focal_class.modifier: @Getter
      @Setter
      private
focal_class.original_string: @Getter
      @Setter
      private int ingestionPoolSize;
focal_class.type: int
focal_class.var_name: ingestionPoolSize
focal_class.declarator: producer
focal_class.modifier: @Getter
      @Setter
      private
focal_class.original_string: @Getter
      @Setter
      private Producer<String, String> producer;
focal_class.type: Producer<String, String>
focal_class.var_name: producer
focal_class.declarator: isTopicOnJsonPath = false
focal_class.modifier: @Getter
      @Setter
      private
focal_class.original_string: @Getter
      @Setter
      private boolean isTopicOnJsonPath = false;
focal_class.type: boolean
focal_class.var_name: isTopicOnJsonPath
focal_class.file: fabric-components/kafka-writer/src/main/java/com/olacabs/fabric/processors/kafkawriter/KafkaWriter.java
focal_class.identifier: KafkaWriter
focal_class.class_method_signature: KafkaWriter.consume(ProcessingContext processingContext, EventSet eventSet)
focal_class.full_signature: @Override protected EventSet consume(ProcessingContext processingContext, EventSet eventSet)
focal_class.identifier: consume
focal_class.modifiers: @Override protected
focal_class.parameters: (ProcessingContext processingContext, EventSet eventSet)
focal_class.return: EventSet
focal_class.signature: EventSet consume(ProcessingContext processingContext, EventSet eventSet)
focal_class.class_method_signature: KafkaWriter.convertEvent(Event event)
focal_class.full_signature: protected KeyedMessage<String, String> convertEvent(Event event)
focal_class.identifier: convertEvent
focal_class.modifiers: protected
focal_class.parameters: (Event event)
focal_class.return: KeyedMessage<String, String>
focal_class.signature: KeyedMessage<String, String> convertEvent(Event event)
focal_class.class_method_signature: KafkaWriter.initialize(String instanceId, Properties globalProperties, Properties properties,
              ComponentMetadata componentMetadata)
focal_class.full_signature: @Override public void initialize(String instanceId, Properties globalProperties, Properties properties,
              ComponentMetadata componentMetadata)
focal_class.identifier: initialize
focal_class.modifiers: @Override public
focal_class.parameters: (String instanceId, Properties globalProperties, Properties properties,
              ComponentMetadata componentMetadata)
focal_class.return: void
focal_class.signature: void initialize(String instanceId, Properties globalProperties, Properties properties,
              ComponentMetadata componentMetadata)
focal_class.class_method_signature: KafkaWriter.destroy()
focal_class.full_signature: @Override public void destroy()
focal_class.identifier: destroy
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void destroy()
focal_class.superclass: extends StreamingProcessor
focal_method.body: @Override
      protected EventSet consume(ProcessingContext processingContext, EventSet eventSet) throws ProcessingException {
          final List<KeyedMessage<String, String>> messages = Lists.newArrayList();
          try {
              eventSet.getEvents().forEach(event -> {
                  KeyedMessage<String, String> convertedMessage = null;
                  try {
                      convertedMessage = convertEvent(event);
                  } catch (ProcessingException e) {
                      LOGGER.error("Error converting byte stream to event: ", e);
                      throw new RuntimeException(e);
                  }
                  if (null != convertedMessage) {
                      messages.add(convertedMessage);
                  }
              });
          } catch (final Exception e) {
              LOGGER.error("Error converting byte stream to event: ", e);
              throw new ProcessingException(e);
          }
          Lists.partition(messages, ingestionPoolSize).forEach(messageList -> getProducer().send(messageList));
          return eventSet;
      }
focal_method.class_method_signature: KafkaWriter.consume(ProcessingContext processingContext, EventSet eventSet)
focal_method.full_signature: @Override protected EventSet consume(ProcessingContext processingContext, EventSet eventSet)
focal_method.identifier: consume
focal_method.invocations: newArrayList forEach getEvents convertEvent error add error forEach partition send getProducer
focal_method.modifiers: @Override protected
focal_method.parameters: (ProcessingContext processingContext, EventSet eventSet)
focal_method.return: EventSet
focal_method.signature: EventSet consume(ProcessingContext processingContext, EventSet eventSet)