LOG statement updated from 'debug' to 'warn'

src/main/config/log4j.dev.properties

@@ -1,18 +1,29 @@
-# log4j logging dev configuration.
+# log4j logging configuration.
 
 # root logger.
 log4j.rootLogger=DEBUG, CONSOLE, ROLLINGFILE
 
 log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
-log4j.appender.CONSOLE.Threshold=INFO
+log4j.appender.CONSOLE.Threshold=WARN
 log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
 log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} [%t] (%C:%L) %-5p %m%n
 
 log4j.appender.ROLLINGFILE=org.apache.log4j.RollingFileAppender
-log4j.appender.ROLLINGFILE.Threshold=DEBUG
-log4j.appender.ROLLINGFILE.File=/tmp/secor_dev/logs/secor.log
+log4j.appender.ROLLINGFILE.Threshold=INFO
+log4j.appender.ROLLINGFILE.File=/tmp/secor-me/logs/secor-${secor_group}.log
 # keep log files up to 1G
 log4j.appender.ROLLINGFILE.MaxFileSize=20MB
 log4j.appender.ROLLINGFILE.MaxBackupIndex=50
 log4j.appender.ROLLINGFILE.layout=org.apache.log4j.PatternLayout
 log4j.appender.ROLLINGFILE.layout.ConversionPattern=%d{ISO8601} [%t] (%C:%L) %-5p %m%n
+
+log4j.appender.ROLLINGFILE2=org.apache.log4j.RollingFileAppender
+log4j.appender.ROLLINGFILE2.Threshold=DEBUG
+log4j.appender.ROLLINGFILE2.File=/tmp/secor-me/logs/secor-uploader.log
+# keep log files up to 1G
+log4j.appender.ROLLINGFILE2.MaxFileSize=200MB
+log4j.appender.ROLLINGFILE2.MaxBackupIndex=5
+log4j.appender.ROLLINGFILE2.layout=org.apache.log4j.PatternLayout
+log4j.appender.ROLLINGFILE2.layout.ConversionPattern=%d{ISO8601} [%t] (%C:%L) %-5p %m%n
+
+log4j.logger.com.pinterest.secor.uploader.Uploader=DEBUG, ROLLINGFILE2

src/main/config/secor.common.properties

@@ -18,11 +18,11 @@
 ############
 
 # Regular expression matching names of consumed topics.
-secor.kafka.topic_filter=*
+secor.kafka.topic_filter=replay
 
 # AWS authentication credentials.
-aws.access.key=
-aws.secret.key=
+aws.access.key=${aws.access.key}
+aws.secret.key=${aws.secret.key}
 
 ################
 # END MUST SET #
@@ -128,21 +128,24 @@ monitoring.blacklist.topics=
 statsd.hostport=
 
 # Name of field that contains timestamp for JSON, MessagePack, or Thrift message parser. (1405970352123)
-message.timestamp.name=
+message.timestamp.name=syncts
 
-# Whether the partition file name should also partition by event name/id
-message.partition.byevent=true
-
-# Name of field that contains message/event id/type for JSON, MessagePack, or Thrift message parser. (1405970352123)
-message.event.name=
+# Name of field that contains timestamp for JSON, MessagePack, or Thrift message parser. (1405970352123)
+message.event.name=eid
 
-# Default folder name to be appended if partition by event is set to true
-message.event.mapping.DEFAULT=logs/
+message.event.mapping.ME_SESSION_SUMMARY=ss/
+message.event.mapping.ME_ASER_SCREEN_SUMMARY=as/
+message.event.mapping.ME_LEARNER_ACTIVITY_SUMMARY=las/
+message.event.mapping.ME_LEARNER_PROFICIENCY_SUMMARY=lp/
+message.event.mapping.ME_LEARNER_CONCEPT_RELEVANCE=lcr/
+message.event.mapping.DEFAULT=me/
 
 # Name of field that contains a timestamp, as a date Format, for JSON. (2014-08-07, Jul 23 02:16:57 2005, etc...)
 # Should be used when there is no timestamp in a Long format. Also ignore time zones.
 message.timestamp.input.pattern=yyyy-MM-dd
 
+message.partition.byevent=true
+
 # To enable compression, set this to a valid compression codec implementing 
 # org.apache.hadoop.io.compress.CompressionCodec interface, such as 
 # 'org.apache.hadoop.io.compress.GzipCodec'.

src/main/config/secor.dev.partition.properties

@@ -16,22 +16,22 @@
 include=secor.dev.properties
 
 # Name of the Kafka consumer group.
-secor.kafka.group=secor_partition
+secor.kafka.group=secor-me
 
 # Parser class that extracts s3 partitions from consumed messages.
 secor.message.parser.class=com.pinterest.secor.parser.PatternDateMessageParser
 
 # S3 path where sequence files are stored.
-secor.s3.path=secor_dev
+secor.s3.path=
 
 # Local path where sequence files are stored before they are uploaded to s3.
-secor.local.path=/tmp/secor_dev/message_logs/partition
+secor.local.path=/tmp/secor-me/message_logs/partition
 
 # Port of the Ostrich server.
-ostrich.port=9998
+ostrich.port=9997
 
 # Partition Date Output format. This is used along with PatternDateMessageParser. Defaults to 'yyyy-MM-dd' *New*
-secor.partition.output_dt_format=dd_MM_yyyy
+secor.partition.output_dt_format=yyyy-MM-dd
 
 # Output file pattern excluding prefix. Defaults to topic/partition/generation_kafkaPartition_fmOffset.gz.
 # Available placeholders are 
@@ -41,7 +41,8 @@ secor.partition.output_dt_format=dd_MM_yyyy
 # kafkaPartition - The kafka partition
 # fmOffset - First Message offset in the file.
 # randomHex - A 4 character random hex to append to the file name
-# currentTimestamp - Epoch time
-# currentTime - Time in HH-mm format
+# currentTimestamp - Time of upload in epoch format
+# currentTime - Time of upload in HH-mm format
+# currentDate - Time of upload in YYYYMMDD format
 # folder - Folder to use based on message id map lookup
-secor.s3.output_file_pattern={partition}-{currentTimestamp}.json
+secor.s3.output_file_pattern={partition}-{currentDate}.json

src/main/config/secor.dev.properties

@@ -5,7 +5,7 @@ include=secor.common.properties
 ############
 
 # Name of the s3 bucket where log files are stored.
-secor.s3.bucket=
+secor.s3.bucket=lpdev-ekstep
 
 ################
 # END MUST SET #
@@ -18,8 +18,8 @@ zookeeper.quorum=localhost:2181
 
 # Upload policies.
 # 10K
-secor.max.file.size.bytes=10000
+secor.max.file.size.bytes=1000000000
 # 10 seconds
-secor.max.file.age.seconds=10
+secor.max.file.age.seconds=300
 # Compute the max file age for a topic partition using either the oldest or newest file. Defaults to newest
 secor.max.file.age.policy=oldest

src/main/java/com/pinterest/secor/uploader/Uploader.java

@@ -211,13 +211,13 @@ private void checkTopicPartition(TopicPartition topicPartition) throws Exception
                 LOG.debug("Uploading for: " + topicPartition);
                 uploadFiles(topicPartition);
             } else if (newOffsetCount > lastSeenOffset) {  // && oldOffset < newOffset
-                LOG.debug("last seen offset {} is lower than committed offset count {}. Deleting files in topic {} partition {}",
+                LOG.warn("last seen offset {} is lower than committed offset count {}. Deleting files in topic {} partition {}",
                         lastSeenOffset, newOffsetCount,topicPartition.getTopic(), topicPartition.getPartition());
                 // There was a rebalancing event and someone committed an offset beyond that of the
                 // current message.  We need to delete the local file.
                 mFileRegistry.deleteTopicPartition(topicPartition);
             } else {  // oldOffsetCount < newOffsetCount <= lastSeenOffset
-                LOG.debug("previous committed offset count {} is lower than committed offset {} is lower than or equal to last seen offset {}. " +
+                LOG.warn("previous committed offset count {} is lower than committed offset {} is lower than or equal to last seen offset {}. " +
                                 "Trimming files in topic {} partition {}",
                         oldOffsetCount, newOffsetCount, lastSeenOffset, topicPartition.getTopic(), topicPartition.getPartition());
                 // There was a rebalancing event and someone committed an offset lower than that