[jira] [Updated] (FLUME-3203) Spooling dir source leaks records from a file when a corresponding .COMPLETED file already present

[Umesh Chaudhary (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLUME-3203?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Umesh Chaudhary updated FLUME-3203:
---
Description: 
Here are the steps to reproduce: 

1) Use below config in the flume agent: 


{code:java}
tier1.sources  = source1
tier1.channels = channel1
tier1.sinks= sink1

tier1.channels.channel1.type   = memory
tier1.channels.channel1.capacity = 1000
tier1.channels.channel1.transactionCapacity = 1000

tier1.sinks.sink1.channel  = channel1
tier1.sources.source1.channels = channel1

tier1.sources.source1.type = spooldir
tier1.sources.source1.spoolDir = /root/testSpoolDir
tier1.sources.source1.fileHeader = true

tier1.sinks.sink1.type = hdfs
tier1.sinks.sink1.hdfs.path =/tmp/spoolEvnts
tier1.sinks.sink1.hdfs.filePrefix = events-
{code}

2) When the agent is started with the above config, use below command to move a 
sample text file in spooling dir:


{code:java}
mv Sample-text-file-50kb.txt /home/systest/spoolDir
{code}

agent will start processing the events and output can be seen in HDFS dir: 

{code:java}
$ hdfs dfs -ls /tmp/spoolEvnts | uniq | wc -l
37
{code}

3) Again move same file into spooling dir using below command: 

{code:java}
mv /tmp/Sample-text-file-50kb.txt /home/systest/spoolDir
{code}

This time flume will raise an exception as below but continue processing the 
file again: 



{noformat}
2017-12-21 00:00:27,581 INFO 
org.apache.flume.client.avro.ReliableSpoolingFileEventReader: Preparing to move 
file /home/systest/spoolDir/Sample-text-file-50kb.txt to 
/home/systest/spoolDir/Sample-text-file-50kb.txt.COMPLETED
2017-12-21 00:00:27,582 ERROR org.apache.flume.source.SpoolDirectorySource: 
FATAL: Spool Directory source source1: { spoolDir: /home/systest/spoolDir }: 
Uncaught exception in SpoolDirectorySource thread. Restart or reconfigure Flume 
to continue processing.
java.lang.IllegalStateException: File name has been re-used with different 
files. Spooling assumptions violated for 
/home/systest/spoolDir/Sample-text-file-50kb.txt.COMPLETED
at 
org.apache.flume.client.avro.ReliableSpoolingFileEventReader.rollCurrentFile(ReliableSpoolingFileEventReader.java:463)
at 
org.apache.flume.client.avro.ReliableSpoolingFileEventReader.retireCurrentFile(ReliableSpoolingFileEventReader.java:414)
at 
org.apache.flume.client.avro.ReliableSpoolingFileEventReader.readEvents(ReliableSpoolingFileEventReader.java:326)
at 
org.apache.flume.source.SpoolDirectorySource$SpoolDirectoryRunnable.run(SpoolDirectorySource.java:250)
at 
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
at 
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
2017-12-21 00:00:31,265 INFO org.apache.flume.sink.hdfs.BucketWriter: Closing 
/tmp/spoolEvnts/events-.1513843202836.tmp
2017-12-21 00:00:31,275 INFO org.apache.flume.sink.hdfs.BucketWriter: Renaming 
/tmp/spoolEvnts/events-.1513843202836.tmp to 
/tmp/spoolEvnts/events-.1513843202836
2017-12-21 00:00:31,293 INFO org.apache.flume.sink.hdfs.BucketWriter: Creating 
/tmp/spoolEvnts/events-.1513843202837.tmp
2017-12-21 00:00:31,321 INFO org.apache.flume.sink.hdfs.BucketWriter: Closing 
/tmp/spoolEvnts/events-.1513843202837.tmp
{noformat}


And if we check at HDFS it shows the below file count :

{code:java}
$ hdfs dfs -ls /tmp/spoolEvnts | uniq | wc -l
72
{code}

Based on [the 
doc|https://flume.apache.org/FlumeUserGuide.html#spooling-directory-source] it 
should not process the file which has same name with .COMPLETED suffix. It 
causes duplicate records on sink.

  was:
Here are the steps to reproduce: 

1) Use below config in the flume agent: 


{code:java}
tier1.sources  = source1
tier1.channels = channel1
tier1.sinks= sink1

tier1.channels.channel1.type   = memory
tier1.channels.channel1.capacity = 1000
tier1.channels.channel1.transactionCapacity = 1000

tier1.sinks.sink1.channel  = channel1
tier1.sources.source1.channels = channel1

tier1.sources.source1.type = spooldir
tier1.sources.source1.spoolDir = /root/testSpoolDir
tier1.sources.source1.fileHeader = true

tier1.sinks.sink1.type = hdfs
tier1.sinks.sink1.hdfs.path =/tmp/spoolEvnts
tier1.sinks.sink1.hdfs.filePrefix = events-
{code}

2) When the agent is started with the above config, use below command to move a 
sample text file in spooling dir:


{code:java}
mv Sample-text-file-50kb.txt /home/systest/spoolDir
{code}

agent 

[jira] [Updated] (FLUME-3203) Spooling dir source leaks records from a file when a corresponding .COMPLETED file already present

[Umesh Chaudhary (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLUME-3203?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Umesh Chaudhary updated FLUME-3203:
---
Description: 
Here are the steps to reproduce: 

1) Use below config in the flume agent: 


{code:java}
tier1.sources  = source1
tier1.channels = channel1
tier1.sinks= sink1

tier1.channels.channel1.type   = memory
tier1.channels.channel1.capacity = 1000
tier1.channels.channel1.transactionCapacity = 1000

tier1.sinks.sink1.channel  = channel1
tier1.sources.source1.channels = channel1

tier1.sources.source1.type = spooldir
tier1.sources.source1.spoolDir = /root/testSpoolDir
tier1.sources.source1.fileHeader = true

tier1.sinks.sink1.type = hdfs
tier1.sinks.sink1.hdfs.path =/tmp/spoolEvnts
tier1.sinks.sink1.hdfs.filePrefix = events-
{code}

2) When the agent is started with the above config, use below command to move a 
sample text file in spooling dir:


{code:java}
mv Sample-text-file-50kb.txt /home/systest/spoolDir
{code}

agent will start processing the events and output can be seen in HDFS dir: 

{code:java}
$ hdfs dfs -ls /tmp/spoolEvnts | uniq | wc -l
37
{code}

3) Again move same file into spooling dir using below command: 

{code:java}
mv /tmp/Sample-text-file-50kb.txt /home/systest/spoolDir
{code}

This time flume will raise an exception as below but continue processing the 
file again: 



{noformat}
2017-12-21 00:00:27,581 INFO 
org.apache.flume.client.avro.ReliableSpoolingFileEventReader: Preparing to move 
file /home/systest/spoolDir/Sample-text-file-50kb.txt to 
/home/systest/spoolDir/Sample-text-file-50kb.txt.COMPLETED
2017-12-21 00:00:27,582 ERROR org.apache.flume.source.SpoolDirectorySource: 
FATAL: Spool Directory source source1: { spoolDir: /home/systest/spoolDir }: 
Uncaught exception in SpoolDirectorySource thread. Restart or reconfigure Flume 
to continue processing.
java.lang.IllegalStateException: File name has been re-used with different 
files. Spooling assumptions violated for 
/home/systest/spoolDir/Sample-text-file-50kb.txt.COMPLETED
at 
org.apache.flume.client.avro.ReliableSpoolingFileEventReader.rollCurrentFile(ReliableSpoolingFileEventReader.java:463)
at 
org.apache.flume.client.avro.ReliableSpoolingFileEventReader.retireCurrentFile(ReliableSpoolingFileEventReader.java:414)
at 
org.apache.flume.client.avro.ReliableSpoolingFileEventReader.readEvents(ReliableSpoolingFileEventReader.java:326)
at 
org.apache.flume.source.SpoolDirectorySource$SpoolDirectoryRunnable.run(SpoolDirectorySource.java:250)
at 
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
at 
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
2017-12-21 00:00:31,265 INFO org.apache.flume.sink.hdfs.BucketWriter: Closing 
/tmp/spoolEvnts/events-.1513843202836.tmp
2017-12-21 00:00:31,275 INFO org.apache.flume.sink.hdfs.BucketWriter: Renaming 
/tmp/spoolEvnts/events-.1513843202836.tmp to 
/tmp/spoolEvnts/events-.1513843202836
2017-12-21 00:00:31,293 INFO org.apache.flume.sink.hdfs.BucketWriter: Creating 
/tmp/spoolEvnts/events-.1513843202837.tmp
2017-12-21 00:00:31,321 INFO org.apache.flume.sink.hdfs.BucketWriter: Closing 
/tmp/spoolEvnts/events-.1513843202837.tmp
{noformat}


And if we check at HDFS it shows the below file count :

{code:java}
$ hdfs dfs -ls /tmp/spoolEvnts | uniq | wc -l
72
{code}

Based on [the 
doc|https://flume.apache.org/FlumeUserGuide.html#spooling-directory-source] it 
should not process the same file which has same name with .COMPLETED suffix. It 
causes duplicate records on sink.

  was:
Here is the steps to reproduce: 

1) Use below config in the flume agent: 


{code:java}
tier1.sources  = source1
tier1.channels = channel1
tier1.sinks= sink1

tier1.channels.channel1.type   = memory
tier1.channels.channel1.capacity = 1000
tier1.channels.channel1.transactionCapacity = 1000

tier1.sinks.sink1.channel  = channel1
tier1.sources.source1.channels = channel1

tier1.sources.source1.type = spooldir
tier1.sources.source1.spoolDir = /root/testSpoolDir
tier1.sources.source1.fileHeader = true

tier1.sinks.sink1.type = hdfs
tier1.sinks.sink1.hdfs.path =/tmp/spoolEvnts
tier1.sinks.sink1.hdfs.filePrefix = events-
{code}

2) When the agent is started with the above config, use below command to move a 
sample text file in spooling dir:


{code:java}
mv Sample-text-file-50kb.txt /home/systest/spoolDir
{code}

agent 

[jira] [Created] (FLUME-3203) Spooling dir source leaks records from a file when a corresponding .COMPLETED file already present

[Umesh Chaudhary (JIRA)]

Umesh Chaudhary created FLUME-3203:
--

 Summary: Spooling dir source leaks records from a file when a 
corresponding .COMPLETED file already present
 Key: FLUME-3203
 URL: https://issues.apache.org/jira/browse/FLUME-3203
 Project: Flume
  Issue Type: Bug
  Components: Sinks+Sources
Affects Versions: 1.6.0
Reporter: Umesh Chaudhary


Here is the steps to reproduce: 

1) Use below config in the flume agent: 


{code:java}
tier1.sources  = source1
tier1.channels = channel1
tier1.sinks= sink1

tier1.channels.channel1.type   = memory
tier1.channels.channel1.capacity = 1000
tier1.channels.channel1.transactionCapacity = 1000

tier1.sinks.sink1.channel  = channel1
tier1.sources.source1.channels = channel1

tier1.sources.source1.type = spooldir
tier1.sources.source1.spoolDir = /root/testSpoolDir
tier1.sources.source1.fileHeader = true

tier1.sinks.sink1.type = hdfs
tier1.sinks.sink1.hdfs.path =/tmp/spoolEvnts
tier1.sinks.sink1.hdfs.filePrefix = events-
{code}

2) When the agent is started with the above config, use below command to move a 
sample text file in spooling dir:


{code:java}
mv Sample-text-file-50kb.txt /home/systest/spoolDir
{code}

agent will start processing the events and output can be seen in HDFS dir: 

{code:java}
$ hdfs dfs -ls /tmp/spoolEvnts | uniq | wc -l
37
{code}

3) Again move same file into spooling dir using below command: 

{code:java}
mv /tmp/Sample-text-file-50kb.txt /home/systest/spoolDir
{code}

This time flume will raise an exception as below but continue processing the 
file again: 



{noformat}
2017-12-21 00:00:27,581 INFO 
org.apache.flume.client.avro.ReliableSpoolingFileEventReader: Preparing to move 
file /home/systest/spoolDir/Sample-text-file-50kb.txt to 
/home/systest/spoolDir/Sample-text-file-50kb.txt.COMPLETED
2017-12-21 00:00:27,582 ERROR org.apache.flume.source.SpoolDirectorySource: 
FATAL: Spool Directory source source1: { spoolDir: /home/systest/spoolDir }: 
Uncaught exception in SpoolDirectorySource thread. Restart or reconfigure Flume 
to continue processing.
java.lang.IllegalStateException: File name has been re-used with different 
files. Spooling assumptions violated for 
/home/systest/spoolDir/Sample-text-file-50kb.txt.COMPLETED
at 
org.apache.flume.client.avro.ReliableSpoolingFileEventReader.rollCurrentFile(ReliableSpoolingFileEventReader.java:463)
at 
org.apache.flume.client.avro.ReliableSpoolingFileEventReader.retireCurrentFile(ReliableSpoolingFileEventReader.java:414)
at 
org.apache.flume.client.avro.ReliableSpoolingFileEventReader.readEvents(ReliableSpoolingFileEventReader.java:326)
at 
org.apache.flume.source.SpoolDirectorySource$SpoolDirectoryRunnable.run(SpoolDirectorySource.java:250)
at 
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
at 
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
at 
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
2017-12-21 00:00:31,265 INFO org.apache.flume.sink.hdfs.BucketWriter: Closing 
/tmp/spoolEvnts/events-.1513843202836.tmp
2017-12-21 00:00:31,275 INFO org.apache.flume.sink.hdfs.BucketWriter: Renaming 
/tmp/spoolEvnts/events-.1513843202836.tmp to 
/tmp/spoolEvnts/events-.1513843202836
2017-12-21 00:00:31,293 INFO org.apache.flume.sink.hdfs.BucketWriter: Creating 
/tmp/spoolEvnts/events-.1513843202837.tmp
2017-12-21 00:00:31,321 INFO org.apache.flume.sink.hdfs.BucketWriter: Closing 
/tmp/spoolEvnts/events-.1513843202837.tmp
{noformat}


And if we check at HDFS it shows the below file count :

{code:java}
$ hdfs dfs -ls /tmp/spoolEvnts | uniq | wc -l
72
{code}

Based on [the 
doc|https://flume.apache.org/FlumeUserGuide.html#spooling-directory-source] it 
should not process the same file which has same name with .COMPLETED suffix. It 
causes duplicate records on sink.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (FLUME-1302) FileChannel Log error message when there is no write permission

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-1302?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15964079#comment-15964079
 ] 

Umesh Chaudhary commented on FLUME-1302:


[~mubarakseyed] after providing the dir path (for checkpoint and data) the only 
reason to print the above message "CheckpointDir could not be created" is 
permission issue. For me the message tells it all, otherwise we need to print 
different message for both the preconditions mentioned in the code. Let me know 
your thoughts. 

> FileChannel Log error message when there is no write permission
> ---
>
> Key: FLUME-1302
> URL: https://issues.apache.org/jira/browse/FLUME-1302
> Project: Flume
>  Issue Type: Improvement
>  Components: Sinks+Sources
>Reporter: Mubarak Seyed
>Priority: Minor
>
> FileChannel Log instance tries to create files (checkpoint and data) and 
> fails when specified directory for flume user (or user one that used to start 
> flume agent) doesn't have write permission. It would be fine if we improve 
> the error message at
> {code}
> Preconditions.checkArgument(
> checkpointDir.isDirectory() || checkpointDir.mkdirs(), "CheckpointDir 
> "
> + checkpointDir + " could not be created");
> {code}
> {code}
> 2012-06-21 09:01:34,530 INFO file.FileChannel: Starting FileChannel with 
> dataDir [/data7/flume/file-channel/data6]
> 2012-06-21 09:01:34,530 ERROR lifecycle.LifecycleSupervisor: Unable to start 
> org.apache.flume.channel.file.FileChannel@6ca084f8 - Exception follows.
> java.lang.IllegalArgumentException: CheckpointDir 
> /data7/flume/file-channel/checkpoint6 could not be created
>   at 
> com.google.common.base.Preconditions.checkArgument(Preconditions.java:88)
>   at org.apache.flume.channel.file.Log.(Log.java:84)
>   at org.apache.flume.channel.file.FileChannel.start(FileChannel.java:178)
>   at 
> org.apache.flume.lifecycle.LifecycleSupervisor$MonitorRunnable.run(LifecycleSupervisor.java:228)
>   at 
> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:441)
>   at 
> java.util.concurrent.FutureTask$Sync.innerRunAndReset(FutureTask.java:317)
>   at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:150)
>   at 
> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$101(ScheduledThreadPoolExecutor.java:98)
>   at 
> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.runPeriodic(ScheduledThreadPoolExecutor.java:180)
>   at 
> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:204)
>   at 
> java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
>   at 
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
>   at java.lang.Thread.run(Thread.java:662)
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLUME-3052) Getting heap space error even changing the memory channel to file channel

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-3052?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15853420#comment-15853420
 ] 

Umesh Chaudhary commented on FLUME-3052:


Hi [~sriramk], Can you attach full stack trace of the error message. Probably 
you should post this query on u...@flume.apache.org. 

> Getting heap space error even changing the memory channel to file channel 
> --
>
> Key: FLUME-3052
> URL: https://issues.apache.org/jira/browse/FLUME-3052
> Project: Flume
>  Issue Type: Bug
>  Components: File Channel, Sinks+Sources
>Reporter: sri
>
>  INFO | Component type: SINK, name: filesink started
> ERROR | Unexpected error processing events
> java.lang.OutOfMemoryError: Java heap space



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (FLUME-1798) Avro source can't bind to wildcard address. Problem with ipv6

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-1798?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15853105#comment-15853105
 ] 

Umesh Chaudhary commented on FLUME-1798:


One more observation regarding this issue. On Ubuntu 14.04, netcat source 
connects via IPv6 protocol instead of the IPv4 protocol.
The config was having following props: 

agent.sources.source-netcat.bind = localhost
agent.sources.source-netcat.port = 5000

There may be multiple reasons for IPv6 precedence over IPv4 as IPv6 was enabled 
on hosts. However using following config fixed the issue: 
 
agent.sources.source-netcat.bind = 0.0.0.0
agent.sources.source-netcat.port = 5000

> Avro source can't bind to wildcard address. Problem with ipv6
> -
>
> Key: FLUME-1798
> URL: https://issues.apache.org/jira/browse/FLUME-1798
> Project: Flume
>  Issue Type: Bug
>  Components: Sinks+Sources
>Affects Versions: v1.3.0
> Environment: Server: Ubuntu 12.04.1 x86_64 (with IPv6), OpenJDK 
> 64-bit IcedTea7 2.3.3
> Client: WinXP (no IPv6), Oracle Java JRE build 1.7.0_05-b05
>Reporter: Greg Bowering
>Priority: Minor
>  Labels: AvroSource, avro
>   Original Estimate: 1h
>  Remaining Estimate: 1h
>
> Thanks for developing Flume-NG :)
> However requiring bind address for AvroSource seems problematic here!
> On WinXP I only have IPv4 available, and bind address same as the hostname 
> works just fine.
> When I migrated my Flume agent to Ubuntu, this no longer works since binding 
> to the server hostname by default seems to only accept incoming IPv6 
> connections.  This breaks for me since all my clients only have IPv4.
> I found I could force AvroSource to accept IPv4 only by binding using the 
> IPv4 address in the conf rather than the hostname.
> This is unsatisfactory since in many cases servers have dynamically assigned 
> IP addresses.
> Furthermore, in some cluster deployment scenarios, even hostnames are not 
> always static (e.g. standing up an EC2 spot-instance to process logs).  It 
> would be much simpler to configure a cluster of Flume agents without having 
> to insert the hostname or IP address in all the conf files, i.e. allow 
> AvroSource to bind to the wildcard address.  This is a simple fix (omit the 
> {{Configurables.ensureRequiredNonNull}} on {{BIND_KEY}} in 
> {{AvroSource#configure(Context)}})
> If sysadmins wish to lock-down network interfaces then there are 
> firewalls/iptables for that kind of stuff...



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Created] (FLUME-3051) Mention the incompatibility of Kafka source with 0.8.x Kafka brokers

[Umesh Chaudhary (JIRA)]

Umesh Chaudhary created FLUME-3051:
--

 Summary: Mention the incompatibility of Kafka source with 0.8.x 
Kafka brokers
 Key: FLUME-3051
 URL: https://issues.apache.org/jira/browse/FLUME-3051
 Project: Flume
  Issue Type: Documentation
  Components: Docs
Affects Versions: v1.7.0
Reporter: Umesh Chaudhary
Assignee: Umesh Chaudhary


In the doc, under Kafka Sink section we mentioned following lines: 

"This currently supports Kafka 0.9.x series of releases.
This version of Flume no longer supports Older Versions (0.8.x) of Kafka."

We need to mention the same for Kafka source as kafka consumer is also using 
new kafka API (0.9.x) 



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Assigned] (FLUME-3023) {variable} substitution doesn't work for property 'fileSuffix'

[Umesh Chaudhary (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLUME-3023?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Umesh Chaudhary reassigned FLUME-3023:
--

Assignee: Umesh Chaudhary

> {variable} substitution doesn't work for property 'fileSuffix'
> --
>
> Key: FLUME-3023
> URL: https://issues.apache.org/jira/browse/FLUME-3023
> Project: Flume
>  Issue Type: Bug
>  Components: Configuration
>Affects Versions: v1.6.0
> Environment: CDH5.5.1
> So, using Flume 1.6.0-cdh5.5.1
>Reporter: Karishma Gulati
>Assignee: Umesh Chaudhary
>Priority: Minor
>
> In my sink configuration, setting the value of fileSuffix as follows :
> {code}
> IngestAgent.sinks.HDFS1.hdfs.fileSuffix=_%Y%m%d.avro
> {code}
> is giving me files with names as follows :
> {code}
> /data/../../MyFileName.1479204507148_%Y%m%d.avro
> {code}
> So the variables %Y%m%d didn't get resolved. 
> Works fine if I add the variables to filePrefix instead. 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLUME-2989) Kafka Channel metrics missing eventTakeAttemptCount and eventPutAttemptCount

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2989?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15599327#comment-15599327
 ] 

Umesh Chaudhary commented on FLUME-2989:


@bessbd Thanks for reviewing the patch.
I resubmitted one after making the suggested changes. 
Please review the updated patch and let me know your thoughts.

> Kafka Channel metrics missing eventTakeAttemptCount and eventPutAttemptCount
> 
>
> Key: FLUME-2989
> URL: https://issues.apache.org/jira/browse/FLUME-2989
> Project: Flume
>  Issue Type: Bug
>  Components: Kafka Channel
>Affects Versions: v1.5.0, v1.6.0, v1.5.1
>Reporter: Denes Arvay
>Assignee: Umesh Chaudhary
>Priority: Minor
> Attachments: FLUME-2989-0.patch, FLUME-2989-1.patch
>
>
> {{eventTakeAttemptCount}} and {{eventPutAttemptCount}} counters don't get 
> incremented in Kafka Channel



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (FLUME-2980) Automated concurrent Kafka offset migration test

[Umesh Chaudhary (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLUME-2980?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Umesh Chaudhary updated FLUME-2980:
---
Attachment: FLUME-2980-0.patch

> Automated concurrent Kafka offset migration test
> 
>
> Key: FLUME-2980
> URL: https://issues.apache.org/jira/browse/FLUME-2980
> Project: Flume
>  Issue Type: Bug
>  Components: Kafka Channel
>Affects Versions: v1.6.0
>Reporter: Mike Percy
>Assignee: Umesh Chaudhary
> Attachments: FLUME-2980-0.patch
>
>
> The Kafka Channel needs an automated offset migration test. See also 
> FLUME-2972



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (FLUME-2989) Kafka Channel metrics missing eventTakeAttemptCount and eventPutAttemptCount

[Umesh Chaudhary (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLUME-2989?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Umesh Chaudhary updated FLUME-2989:
---
Attachment: FLUME-2989-1.patch

> Kafka Channel metrics missing eventTakeAttemptCount and eventPutAttemptCount
> 
>
> Key: FLUME-2989
> URL: https://issues.apache.org/jira/browse/FLUME-2989
> Project: Flume
>  Issue Type: Bug
>  Components: Kafka Channel
>Affects Versions: v1.5.0, v1.6.0, v1.5.1
>Reporter: Denes Arvay
>Assignee: Umesh Chaudhary
>Priority: Minor
> Attachments: FLUME-2989-0.patch, FLUME-2989-1.patch
>
>
> {{eventTakeAttemptCount}} and {{eventPutAttemptCount}} counters don't get 
> incremented in Kafka Channel



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (FLUME-2989) Kafka Channel metrics missing eventTakeAttemptCount and eventPutAttemptCount

[Umesh Chaudhary (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLUME-2989?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Umesh Chaudhary updated FLUME-2989:
---
Attachment: FLUME-2989-0.patch

> Kafka Channel metrics missing eventTakeAttemptCount and eventPutAttemptCount
> 
>
> Key: FLUME-2989
> URL: https://issues.apache.org/jira/browse/FLUME-2989
> Project: Flume
>  Issue Type: Bug
>  Components: Kafka Channel
>Affects Versions: v1.5.0, v1.6.0, v1.5.1
>Reporter: Denes Arvay
>Assignee: Umesh Chaudhary
>Priority: Minor
> Attachments: FLUME-2989-0.patch
>
>
> {{eventTakeAttemptCount}} and {{eventPutAttemptCount}} counters don't get 
> incremented in Kafka Channel



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Assigned] (FLUME-2980) Automated concurrent Kafka offset migration test

[Umesh Chaudhary (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLUME-2980?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Umesh Chaudhary reassigned FLUME-2980:
--

Assignee: Umesh Chaudhary

> Automated concurrent Kafka offset migration test
> 
>
> Key: FLUME-2980
> URL: https://issues.apache.org/jira/browse/FLUME-2980
> Project: Flume
>  Issue Type: Bug
>  Components: Kafka Channel
>Reporter: Mike Percy
>Assignee: Umesh Chaudhary
>
> The Kafka Channel needs an automated offset migration test. See also 
> FLUME-2972



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLUME-2980) Automated concurrent Kafka offset migration test

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2980?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15542519#comment-15542519
 ] 

Umesh Chaudhary commented on FLUME-2980:


Hi [~mpercy], I would like to work on this.

> Automated concurrent Kafka offset migration test
> 
>
> Key: FLUME-2980
> URL: https://issues.apache.org/jira/browse/FLUME-2980
> Project: Flume
>  Issue Type: Bug
>  Components: Kafka Channel
>Reporter: Mike Percy
>
> The Kafka Channel needs an automated offset migration test. See also 
> FLUME-2972



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLUME-2623) Test testRemoveFields(org.apache.flume.source.TestSyslogUdpSource) fails randomly

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2623?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15535448#comment-15535448
 ] 

Umesh Chaudhary commented on FLUME-2623:


Thanks [~bessbd]. I will take it start working on this.

> Test testRemoveFields(org.apache.flume.source.TestSyslogUdpSource) fails 
> randomly
> -
>
> Key: FLUME-2623
> URL: https://issues.apache.org/jira/browse/FLUME-2623
> Project: Flume
>  Issue Type: Bug
>  Components: Test
>Affects Versions: v1.5.0
> Environment: RHEL 7.1 on PPC64 LE
>Reporter: Tony Reix
>Priority: Minor
>
> The test org.apache.flume.source.TestSyslogUdpSource is not 100% reliable. It 
> fails sometimes randomly.
> Source code dealing with the issue is:
> flume-ng-core/src/test/java/org/apache/flume/source/TestSyslogUdpSource.java  
> (about line 101)
> for (int i = 0; i < 100 ; i++) {
>   syslogSocket = new DatagramSocket();
>   syslogSocket.send(datagramPacket);
>   syslogSocket.close();
> }
> List channelEvents = new ArrayList();
> Transaction txn = channel.getTransaction();
> txn.begin();
> for (int i = 0; i < 100; i++) {
>   Event e = channel.take();
>   Assert.assertNotNull(e);
>   channelEvents.add(e);
> }
> Sometimes... "e" is null.
> Failure deals with:
> at org.junit.Assert.assertTrue(Assert.java:43)
> at org.junit.Assert.assertNotNull(Assert.java:526) at 
> org.junit.Assert.assertNotNull(Assert.java:537)
> at 
> org.apache.flume.source.TestSyslogUdpSource.runKeepFieldsTest(TestSyslogUdpSource.java:101)
> at 
> org.apache.flume.source.TestSyslogUdpSource.testRemoveFields(TestSyslogUdpSource.java:177)
> With OpenJDK, I got it failing once out of 30 tries.
> However, with IBM JVM, I got if failing 6 times out of 10.
> After I added a Thread.sleep(2000) in the middle, with IBM JVM, I've reduced 
> the probability of the failure from 6/10 to 2/10 . So, that helps, but that 
> is not enough. A better solution must be found.
> The issue appears more often with IBM JVM probably because things are handled 
> differently, or quicker, by IBM JVM. Anyway, the issue also appears with 
> OpenJDK.
> I guess that the issue is still there with master version.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Comment Edited] (FLUME-2623) Test testRemoveFields(org.apache.flume.source.TestSyslogUdpSource) fails randomly

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2623?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15535448#comment-15535448
 ] 

Umesh Chaudhary edited comment on FLUME-2623 at 9/30/16 8:49 AM:
-

Thanks [~bessbd]. I will take it and start working on this.


was (Author: umesh9...@gmail.com):
Thanks [~bessbd]. I will take it start working on this.

> Test testRemoveFields(org.apache.flume.source.TestSyslogUdpSource) fails 
> randomly
> -
>
> Key: FLUME-2623
> URL: https://issues.apache.org/jira/browse/FLUME-2623
> Project: Flume
>  Issue Type: Bug
>  Components: Test
>Affects Versions: v1.5.0
> Environment: RHEL 7.1 on PPC64 LE
>Reporter: Tony Reix
>Priority: Minor
>
> The test org.apache.flume.source.TestSyslogUdpSource is not 100% reliable. It 
> fails sometimes randomly.
> Source code dealing with the issue is:
> flume-ng-core/src/test/java/org/apache/flume/source/TestSyslogUdpSource.java  
> (about line 101)
> for (int i = 0; i < 100 ; i++) {
>   syslogSocket = new DatagramSocket();
>   syslogSocket.send(datagramPacket);
>   syslogSocket.close();
> }
> List channelEvents = new ArrayList();
> Transaction txn = channel.getTransaction();
> txn.begin();
> for (int i = 0; i < 100; i++) {
>   Event e = channel.take();
>   Assert.assertNotNull(e);
>   channelEvents.add(e);
> }
> Sometimes... "e" is null.
> Failure deals with:
> at org.junit.Assert.assertTrue(Assert.java:43)
> at org.junit.Assert.assertNotNull(Assert.java:526) at 
> org.junit.Assert.assertNotNull(Assert.java:537)
> at 
> org.apache.flume.source.TestSyslogUdpSource.runKeepFieldsTest(TestSyslogUdpSource.java:101)
> at 
> org.apache.flume.source.TestSyslogUdpSource.testRemoveFields(TestSyslogUdpSource.java:177)
> With OpenJDK, I got it failing once out of 30 tries.
> However, with IBM JVM, I got if failing 6 times out of 10.
> After I added a Thread.sleep(2000) in the middle, with IBM JVM, I've reduced 
> the probability of the failure from 6/10 to 2/10 . So, that helps, but that 
> is not enough. A better solution must be found.
> The issue appears more often with IBM JVM probably because things are handled 
> differently, or quicker, by IBM JVM. Anyway, the issue also appears with 
> OpenJDK.
> I guess that the issue is still there with master version.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLUME-2883) LoadBalancingRpcClient returns wrong value for batchSize

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2883?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15535071#comment-15535071
 ] 

Umesh Chaudhary commented on FLUME-2883:


Hi [~agnek], Apologies as I assign this to me without asking to you. Would you 
want to work on this ?

> LoadBalancingRpcClient returns wrong value for batchSize
> 
>
> Key: FLUME-2883
> URL: https://issues.apache.org/jira/browse/FLUME-2883
> Project: Flume
>  Issue Type: Bug
>  Components: Client SDK
>Reporter: Alexey Churkin
>Assignee: Umesh Chaudhary
>Priority: Minor
>
> LoadBalancingRpcClient doesn't read batchSize property from configuration and 
> returns default value in getBatchSize method.
> But underlying clients creates with right batchSize value read from 
> configuration. So LoadBalancingRpcClient should also read batch-size 
> configuration param and set it field batchSize according to it value.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLUME-2811) Taildir source doesn't call stop() on graceful shutdown

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2811?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15535050#comment-15535050
 ] 

Umesh Chaudhary commented on FLUME-2811:


Hello [~siefried12], [~liorze] IMHO InterruptedException would be thrown when 
the thread is interrupted in method tailFileProcess and ChannelException is 
thrown when channel is misbehaving. So InterruptedException should not cause 
infinite loop. 

> Taildir source doesn't call stop() on graceful shutdown
> ---
>
> Key: FLUME-2811
> URL: https://issues.apache.org/jira/browse/FLUME-2811
> Project: Flume
>  Issue Type: Bug
>  Components: Sinks+Sources
>Affects Versions: v1.7.0
>Reporter: Jun Seok Hong
>Assignee: Umesh Chaudhary
>Priority: Critical
>  Labels: newbie
> Fix For: v1.7.0
>
>
> Taildir source doesn't call stop() on graceful shutdown.
> Test configuration.
> source - taildir
> channel - PseudoTxnMemoryChannel / flume-kafka-channel
> sink - none
> I found that flume sometimes doesn't terminate with Taildir source. 
> I had to kill the process to terminate it.
> tailFileProcess() function in TaildirSource.java has a infinite loop.
> When the process interrupted, ChannelException will happen, but it can't 
> breaks the infinite loop.
> I think that's the reason why Taildir can't call stop() function.
> {code:title=TaildirSource.java|borderStyle=solid}
>  private void tailFileProcess(TailFile tf, boolean backoffWithoutNL)
>   throws IOException, InterruptedException {
> while (true) {
>   reader.setCurrentFile(tf);
>   List events = reader.readEvents(batchSize, backoffWithoutNL);
>   if (events.isEmpty()) {
> break;
>   }
>   sourceCounter.addToEventReceivedCount(events.size());
>   sourceCounter.incrementAppendBatchReceivedCount();
>   try {
> getChannelProcessor().processEventBatch(events);
> reader.commit();
>   } catch (ChannelException ex) {
> logger.warn("The channel is full or unexpected failure. " +
>   "The source will try again after " + retryInterval + " ms");
> TimeUnit.MILLISECONDS.sleep(retryInterval);
> retryInterval = retryInterval << 1;
> retryInterval = Math.min(retryInterval, maxRetryInterval);
> continue;
>   }
>   retryInterval = 1000;
>   sourceCounter.addToEventAcceptedCount(events.size());
>   sourceCounter.incrementAppendBatchAcceptedCount();
>   if (events.size() < batchSize) {
> break;
>   }
> }
>   }
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLUME-2583) Flafka unit tests should randomize ports

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2583?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15535035#comment-15535035
 ] 

Umesh Chaudhary commented on FLUME-2583:


Hi [~gwenshap] , would it be fine if I take this one? Will need your inputs 
though.

> Flafka unit tests should randomize ports
> 
>
> Key: FLUME-2583
> URL: https://issues.apache.org/jira/browse/FLUME-2583
> Project: Flume
>  Issue Type: Improvement
>Reporter: Gwen Shapira
>Assignee: Gwen Shapira
>
> Flafka (i.e. Kafka source, sink and channel) unit tests don't randomize 
> ports, therefore they can fail (after creating a small mess) when running on 
> machines that already have Kafka or Zookeeper running.
> Lets fix this.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLUME-2623) Test testRemoveFields(org.apache.flume.source.TestSyslogUdpSource) fails randomly

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2623?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15535027#comment-15535027
 ] 

Umesh Chaudhary commented on FLUME-2623:


Hi [~hshreedharan] [~jholoman] [~bessbd] Would it be fine if I take this one?

> Test testRemoveFields(org.apache.flume.source.TestSyslogUdpSource) fails 
> randomly
> -
>
> Key: FLUME-2623
> URL: https://issues.apache.org/jira/browse/FLUME-2623
> Project: Flume
>  Issue Type: Bug
>  Components: Test
>Affects Versions: v1.5.0
> Environment: RHEL 7.1 on PPC64 LE
>Reporter: Tony Reix
>Priority: Minor
>
> The test org.apache.flume.source.TestSyslogUdpSource is not 100% reliable. It 
> fails sometimes randomly.
> Source code dealing with the issue is:
> flume-ng-core/src/test/java/org/apache/flume/source/TestSyslogUdpSource.java  
> (about line 101)
> for (int i = 0; i < 100 ; i++) {
>   syslogSocket = new DatagramSocket();
>   syslogSocket.send(datagramPacket);
>   syslogSocket.close();
> }
> List channelEvents = new ArrayList();
> Transaction txn = channel.getTransaction();
> txn.begin();
> for (int i = 0; i < 100; i++) {
>   Event e = channel.take();
>   Assert.assertNotNull(e);
>   channelEvents.add(e);
> }
> Sometimes... "e" is null.
> Failure deals with:
> at org.junit.Assert.assertTrue(Assert.java:43)
> at org.junit.Assert.assertNotNull(Assert.java:526) at 
> org.junit.Assert.assertNotNull(Assert.java:537)
> at 
> org.apache.flume.source.TestSyslogUdpSource.runKeepFieldsTest(TestSyslogUdpSource.java:101)
> at 
> org.apache.flume.source.TestSyslogUdpSource.testRemoveFields(TestSyslogUdpSource.java:177)
> With OpenJDK, I got it failing once out of 30 tries.
> However, with IBM JVM, I got if failing 6 times out of 10.
> After I added a Thread.sleep(2000) in the middle, with IBM JVM, I've reduced 
> the probability of the failure from 6/10 to 2/10 . So, that helps, but that 
> is not enough. A better solution must be found.
> The issue appears more often with IBM JVM probably because things are handled 
> differently, or quicker, by IBM JVM. Anyway, the issue also appears with 
> OpenJDK.
> I guess that the issue is still there with master version.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLUME-2894) Flume components should stop in the correct order (graceful shutdown)

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2894?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15529354#comment-15529354
 ] 

Umesh Chaudhary commented on FLUME-2894:


Wondering if we can achieve this by current data structures used in 
LifecycleSupervisor class. We are using HashMap as:

supervisedProcesses = new HashMap();

And it is having key as LifecycleAware which is not a NamedComponent. [~bessbd] 
[~jholoman], Need your inputs to do this.

> Flume components should stop in the correct order (graceful shutdown)
> -
>
> Key: FLUME-2894
> URL: https://issues.apache.org/jira/browse/FLUME-2894
> Project: Flume
>  Issue Type: Bug
>  Components: Channel, Node, Sinks+Sources
>Affects Versions: v1.6.0
>Reporter: Piotr Wiecek
>
> Flume components should be stopped in the right way:
> * stop the sources (in order to not receiving further notifications),
> * wait until all events within the channels are consumed by the sinks,
> * stop the channels and the sinks.
> Currently, the shutdown hook stops the components in a random manner.
> E.g.: SINK, CHANNEL, SOURCE.
> Components are stored in the HashMap:
> {code:borderStyle=solid}
> Map supervisedProcesses;
> ...
> supervisedProcesses = new HashMap();
> ...
> @Override
>   public synchronized void stop() {
>   ...
>   for (final Entry entry : supervisedProcesses
> .entrySet()) {
>   if (entry.getKey().getLifecycleState().equals(LifecycleState.START)) {
> entry.getValue().status.desiredState = LifecycleState.STOP;
> entry.getKey().stop();
>   }
> }
> 
> {code}
> The problems which we can have:
> * not all Events will be consumed (Sink will be stopped first)
> * Source will continue to accept messages even though other components are 
> stopped



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLUME-2795) Sinks with hdfs path with escape sequence do not close current .tmp file when changit to new directory

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2795?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15528863#comment-15528863
 ] 

Umesh Chaudhary commented on FLUME-2795:


Hi [~dscarlat], do you still see this issue? If so, would you mind explaining 
the scenario bit more and did you try setting hdfs.idleTimeout ?

> Sinks with hdfs path with escape sequence do not close current .tmp file when 
> changit to new directory
> --
>
> Key: FLUME-2795
> URL: https://issues.apache.org/jira/browse/FLUME-2795
> Project: Flume
>  Issue Type: Bug
>  Components: Sinks+Sources
>Affects Versions: v1.5.0
> Environment: cdh5.4.4
>  over ubuntu
>Reporter: David Scarlatti
>
> I have a hdfs sink with this config:
> tier1.sinks.sink1.type = hdfs
> tier1.sinks.sink1.channel  = channel1
> tier1.sinks.sink1.hdfs.path= /user/bla/%y-%m-%d
> tier1.sinks.sink1.hdfs.filePrefix =bla
> tier1.sinks.sink1.hdfs.rollSize = 0
> tier1.sinks.sink1.hdfs.rollInterval = 0
> tier1.sinks.sink1.hdfs.rollCount = 15
> tier1.sinks.sink1.hdfs.useLocalTimeStamp = true
> tier1.sinks.sink1.hdfs.fileType = DataStream
> tier1.sinks.sink1.hdfs.batchSize = 100
> every night at 23:59:59 a new folder is created in the HDFS and the folder 
> for the previous day has a last file with .tmp extension, the file is 
> incomplete and only when the flume agent is restarted this .tmp file is 
> completed and closed an renamed.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Assigned] (FLUME-2883) LoadBalancingRpcClient returns wrong value for batchSize

[Umesh Chaudhary (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLUME-2883?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Umesh Chaudhary reassigned FLUME-2883:
--

Assignee: Umesh Chaudhary

> LoadBalancingRpcClient returns wrong value for batchSize
> 
>
> Key: FLUME-2883
> URL: https://issues.apache.org/jira/browse/FLUME-2883
> Project: Flume
>  Issue Type: Bug
>  Components: Client SDK
>Reporter: Alexey Churkin
>Assignee: Umesh Chaudhary
>Priority: Minor
>
> LoadBalancingRpcClient doesn't read batchSize property from configuration and 
> returns default value in getBatchSize method.
> But underlying clients creates with right batchSize value read from 
> configuration. So LoadBalancingRpcClient should also read batch-size 
> configuration param and set it field batchSize according to it value.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Assigned] (FLUME-2811) Taildir source doesn't call stop() on graceful shutdown

[Umesh Chaudhary (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLUME-2811?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Umesh Chaudhary reassigned FLUME-2811:
--

Assignee: Umesh Chaudhary

> Taildir source doesn't call stop() on graceful shutdown
> ---
>
> Key: FLUME-2811
> URL: https://issues.apache.org/jira/browse/FLUME-2811
> Project: Flume
>  Issue Type: Bug
>  Components: Sinks+Sources
>Affects Versions: v1.7.0
>Reporter: Jun Seok Hong
>Assignee: Umesh Chaudhary
>Priority: Critical
>  Labels: newbie
> Fix For: v1.7.0
>
>
> Taildir source doesn't call stop() on graceful shutdown.
> Test configuration.
> source - taildir
> channel - PseudoTxnMemoryChannel / flume-kafka-channel
> sink - none
> I found that flume sometimes doesn't terminate with Taildir source. 
> I had to kill the process to terminate it.
> tailFileProcess() function in TaildirSource.java has a infinite loop.
> When the process interrupted, ChannelException will happen, but it can't 
> breaks the infinite loop.
> I think that's the reason why Taildir can't call stop() function.
> {code:title=TaildirSource.java|borderStyle=solid}
>  private void tailFileProcess(TailFile tf, boolean backoffWithoutNL)
>   throws IOException, InterruptedException {
> while (true) {
>   reader.setCurrentFile(tf);
>   List events = reader.readEvents(batchSize, backoffWithoutNL);
>   if (events.isEmpty()) {
> break;
>   }
>   sourceCounter.addToEventReceivedCount(events.size());
>   sourceCounter.incrementAppendBatchReceivedCount();
>   try {
> getChannelProcessor().processEventBatch(events);
> reader.commit();
>   } catch (ChannelException ex) {
> logger.warn("The channel is full or unexpected failure. " +
>   "The source will try again after " + retryInterval + " ms");
> TimeUnit.MILLISECONDS.sleep(retryInterval);
> retryInterval = retryInterval << 1;
> retryInterval = Math.min(retryInterval, maxRetryInterval);
> continue;
>   }
>   retryInterval = 1000;
>   sourceCounter.addToEventAcceptedCount(events.size());
>   sourceCounter.incrementAppendBatchAcceptedCount();
>   if (events.size() < batchSize) {
> break;
>   }
> }
>   }
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (FLUME-2647) File Channel must print the reason for the channel being closed in channelClosed checks

[Umesh Chaudhary (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLUME-2647?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Umesh Chaudhary updated FLUME-2647:
---
Attachment: (was: FLUME-2647-0.patch)

> File Channel must print the reason for the channel being closed in 
> channelClosed checks
> ---
>
> Key: FLUME-2647
> URL: https://issues.apache.org/jira/browse/FLUME-2647
> Project: Flume
>  Issue Type: Bug
>Reporter: Hari Shreedharan
>Assignee: Umesh Chaudhary
> Attachments: FLUME-2647-3.patch
>
>
> Before each operation we check if the channel is open, but if it is closed, 
> we don't say how it was closed. We must print this exception to the logs.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (FLUME-2647) File Channel must print the reason for the channel being closed in channelClosed checks

[Umesh Chaudhary (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLUME-2647?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Umesh Chaudhary updated FLUME-2647:
---
Attachment: FLUME-2647-3.patch

> File Channel must print the reason for the channel being closed in 
> channelClosed checks
> ---
>
> Key: FLUME-2647
> URL: https://issues.apache.org/jira/browse/FLUME-2647
> Project: Flume
>  Issue Type: Bug
>Reporter: Hari Shreedharan
>Assignee: Umesh Chaudhary
> Attachments: FLUME-2647-3.patch
>
>
> Before each operation we check if the channel is open, but if it is closed, 
> we don't say how it was closed. We must print this exception to the logs.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Comment Edited] (FLUME-2647) File Channel must print the reason for the channel being closed in channelClosed checks

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2647?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15518851#comment-15518851
 ] 

Umesh Chaudhary edited comment on FLUME-2647 at 9/24/16 10:44 AM:
--

Hi @bessbd, 
Thanks for reviewing the patch.
Sure, will submit the updated patch soon.


was (Author: umesh9...@gmail.com):
Hi @bessbd, 
Sure, will submit the updated patch soon.

> File Channel must print the reason for the channel being closed in 
> channelClosed checks
> ---
>
> Key: FLUME-2647
> URL: https://issues.apache.org/jira/browse/FLUME-2647
> Project: Flume
>  Issue Type: Bug
>Reporter: Hari Shreedharan
>Assignee: Umesh Chaudhary
> Attachments: FLUME-2647-0.patch
>
>
> Before each operation we check if the channel is open, but if it is closed, 
> we don't say how it was closed. We must print this exception to the logs.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLUME-2647) File Channel must print the reason for the channel being closed in channelClosed checks

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2647?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15518851#comment-15518851
 ] 

Umesh Chaudhary commented on FLUME-2647:


Hi @bessbd, 
Sure, will submit the updated patch soon.

> File Channel must print the reason for the channel being closed in 
> channelClosed checks
> ---
>
> Key: FLUME-2647
> URL: https://issues.apache.org/jira/browse/FLUME-2647
> Project: Flume
>  Issue Type: Bug
>Reporter: Hari Shreedharan
>Assignee: Umesh Chaudhary
> Attachments: FLUME-2647-0.patch
>
>
> Before each operation we check if the channel is open, but if it is closed, 
> we don't say how it was closed. We must print this exception to the logs.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (FLUME-2647) File Channel must print the reason for the channel being closed in channelClosed checks

[Umesh Chaudhary (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLUME-2647?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Umesh Chaudhary updated FLUME-2647:
---
Attachment: FLUME-2647-0.patch

> File Channel must print the reason for the channel being closed in 
> channelClosed checks
> ---
>
> Key: FLUME-2647
> URL: https://issues.apache.org/jira/browse/FLUME-2647
> Project: Flume
>  Issue Type: Bug
>Reporter: Hari Shreedharan
>Assignee: Umesh Chaudhary
> Attachments: FLUME-2647-0.patch
>
>
> Before each operation we check if the channel is open, but if it is closed, 
> we don't say how it was closed. We must print this exception to the logs.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLUME-2647) File Channel must print the reason for the channel being closed in channelClosed checks

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2647?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15516300#comment-15516300
 ] 

Umesh Chaudhary commented on FLUME-2647:


Hi [~bessbd] [~jholoman] , I have submitted a review request for my patch on 
Review Board. Please let me know your comments.

> File Channel must print the reason for the channel being closed in 
> channelClosed checks
> ---
>
> Key: FLUME-2647
> URL: https://issues.apache.org/jira/browse/FLUME-2647
> Project: Flume
>  Issue Type: Bug
>Reporter: Hari Shreedharan
>Assignee: Umesh Chaudhary
>
> Before each operation we check if the channel is open, but if it is closed, 
> we don't say how it was closed. We must print this exception to the logs.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Assigned] (FLUME-2647) File Channel must print the reason for the channel being closed in channelClosed checks

[Umesh Chaudhary (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLUME-2647?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Umesh Chaudhary reassigned FLUME-2647:
--

Assignee: Umesh Chaudhary  (was: Johny Rufus)

> File Channel must print the reason for the channel being closed in 
> channelClosed checks
> ---
>
> Key: FLUME-2647
> URL: https://issues.apache.org/jira/browse/FLUME-2647
> Project: Flume
>  Issue Type: Bug
>Reporter: Hari Shreedharan
>Assignee: Umesh Chaudhary
>
> Before each operation we check if the channel is open, but if it is closed, 
> we don't say how it was closed. We must print this exception to the logs.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLUME-2647) File Channel must print the reason for the channel being closed in channelClosed checks

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2647?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15513040#comment-15513040
 ] 

Umesh Chaudhary commented on FLUME-2647:


Hi [~bessbd], actually my email ID is 'umesh9...@gmail.com' . Please add this. 
The earlier one which you've added has almost same name as mine. Thanks !

> File Channel must print the reason for the channel being closed in 
> channelClosed checks
> ---
>
> Key: FLUME-2647
> URL: https://issues.apache.org/jira/browse/FLUME-2647
> Project: Flume
>  Issue Type: Bug
>Reporter: Hari Shreedharan
>Assignee: Johny Rufus
>
> Before each operation we check if the channel is open, but if it is closed, 
> we don't say how it was closed. We must print this exception to the logs.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLUME-2647) File Channel must print the reason for the channel being closed in channelClosed checks

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2647?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15512368#comment-15512368
 ] 

Umesh Chaudhary commented on FLUME-2647:


Thanks [~jholoman], I will start to work on this. May I ask you to assign this 
to me?

> File Channel must print the reason for the channel being closed in 
> channelClosed checks
> ---
>
> Key: FLUME-2647
> URL: https://issues.apache.org/jira/browse/FLUME-2647
> Project: Flume
>  Issue Type: Bug
>Reporter: Hari Shreedharan
>Assignee: Johny Rufus
>
> Before each operation we check if the channel is open, but if it is closed, 
> we don't say how it was closed. We must print this exception to the logs.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLUME-2883) LoadBalancingRpcClient returns wrong value for batchSize

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2883?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15503623#comment-15503623
 ] 

Umesh Chaudhary commented on FLUME-2883:


Yes, that is a valid point. [~mpercy], please provide your inputs on this. 

> LoadBalancingRpcClient returns wrong value for batchSize
> 
>
> Key: FLUME-2883
> URL: https://issues.apache.org/jira/browse/FLUME-2883
> Project: Flume
>  Issue Type: Bug
>  Components: Client SDK
>Reporter: Alexey Churkin
>Priority: Minor
>
> LoadBalancingRpcClient doesn't read batchSize property from configuration and 
> returns default value in getBatchSize method.
> But underlying clients creates with right batchSize value read from 
> configuration. So LoadBalancingRpcClient should also read batch-size 
> configuration param and set it field batchSize according to it value.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLUME-2883) LoadBalancingRpcClient returns wrong value for batchSize

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2883?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15483941#comment-15483941
 ] 

Umesh Chaudhary commented on FLUME-2883:


[~agnek], I am sorry but I don't see any getBatchSize method in 
LoadBalancingRpcClient. 

> LoadBalancingRpcClient returns wrong value for batchSize
> 
>
> Key: FLUME-2883
> URL: https://issues.apache.org/jira/browse/FLUME-2883
> Project: Flume
>  Issue Type: Bug
>  Components: Client SDK
>Reporter: Alexey Churkin
>Priority: Minor
>
> LoadBalancingRpcClient doesn't read batchSize property from configuration and 
> returns default value in getBatchSize method.
> But underlying clients creates with right batchSize value read from 
> configuration. So LoadBalancingRpcClient should also read batch-size 
> configuration param and set it field batchSize according to it value.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLUME-2647) File Channel must print the reason for the channel being closed in channelClosed checks

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2647?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15483864#comment-15483864
 ] 

Umesh Chaudhary commented on FLUME-2647:


Hello [~jrufus], If you want I can work on this.

> File Channel must print the reason for the channel being closed in 
> channelClosed checks
> ---
>
> Key: FLUME-2647
> URL: https://issues.apache.org/jira/browse/FLUME-2647
> Project: Flume
>  Issue Type: Bug
>Reporter: Hari Shreedharan
>Assignee: Johny Rufus
>
> Before each operation we check if the channel is open, but if it is closed, 
> we don't say how it was closed. We must print this exception to the logs.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLUME-2811) Taildir source doesn't call stop() on graceful shutdown

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2811?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15483206#comment-15483206
 ] 

Umesh Chaudhary commented on FLUME-2811:


Hello [~siefried12] [~liorze] , I can work for this.

> Taildir source doesn't call stop() on graceful shutdown
> ---
>
> Key: FLUME-2811
> URL: https://issues.apache.org/jira/browse/FLUME-2811
> Project: Flume
>  Issue Type: Bug
>  Components: Sinks+Sources
>Affects Versions: v1.7.0
>Reporter: Jun Seok Hong
>Priority: Critical
>  Labels: newbie
> Fix For: v1.7.0
>
>
> Taildir source doesn't call stop() on graceful shutdown.
> Test configuration.
> source - taildir
> channel - PseudoTxnMemoryChannel / flume-kafka-channel
> sink - none
> I found that flume sometimes doesn't terminate with Taildir source. 
> I had to kill the process to terminate it.
> tailFileProcess() function in TaildirSource.java has a infinite loop.
> When the process interrupted, ChannelException will happen, but it can't 
> breaks the infinite loop.
> I think that's the reason why Taildir can't call stop() function.
> {code:title=TaildirSource.java|borderStyle=solid}
>  private void tailFileProcess(TailFile tf, boolean backoffWithoutNL)
>   throws IOException, InterruptedException {
> while (true) {
>   reader.setCurrentFile(tf);
>   List events = reader.readEvents(batchSize, backoffWithoutNL);
>   if (events.isEmpty()) {
> break;
>   }
>   sourceCounter.addToEventReceivedCount(events.size());
>   sourceCounter.incrementAppendBatchReceivedCount();
>   try {
> getChannelProcessor().processEventBatch(events);
> reader.commit();
>   } catch (ChannelException ex) {
> logger.warn("The channel is full or unexpected failure. " +
>   "The source will try again after " + retryInterval + " ms");
> TimeUnit.MILLISECONDS.sleep(retryInterval);
> retryInterval = retryInterval << 1;
> retryInterval = Math.min(retryInterval, maxRetryInterval);
> continue;
>   }
>   retryInterval = 1000;
>   sourceCounter.addToEventAcceptedCount(events.size());
>   sourceCounter.incrementAppendBatchAcceptedCount();
>   if (events.size() < batchSize) {
> break;
>   }
> }
>   }
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLUME-2692) Make the number of threads in AsyncHbaseSink configurable

[Umesh Chaudhary (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLUME-2692?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15483188#comment-15483188
 ] 

Umesh Chaudhary commented on FLUME-2692:


Hello [~jrufus], Are you still working on this? If not, I can work on this. 
In my opinion,  there should be some formula to decide Hbase client connection 
threads and that should be well documented. Just like 2 * num_processors, which 
is the simplest way here to decide parallelism. Want to know you thoughts 
though.

I see that we could utilize following constructor:
 public NioClientSocketChannelFactory(Executor bossExecutor, Executor 
workerExecutor, int bossCount, int workerCount) 


> Make the number of threads in AsyncHbaseSink configurable
> -
>
> Key: FLUME-2692
> URL: https://issues.apache.org/jira/browse/FLUME-2692
> Project: Flume
>  Issue Type: Improvement
>Reporter: Johny Rufus
>Assignee: Johny Rufus
>
> When creating the HbaseClient, NioClientSocketChannelFactory passed in, can 
> take a configurable number of threads as parameter. The default number of 
> worker threads created is 2 * num_processors (similar on both client and 
> server side)



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)