WARN conf.FlumeConfiguration: Could not configure sink  k1 due to: Channel c1 not in active set.

org.apache.flume.conf.ConfigurationException: Channel c1 not in active set.

	at org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.validateSinks(FlumeConfiguration.java:685)

	at org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.isValid(FlumeConfiguration.java:347)

	at org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.access$000(FlumeConfiguration.java:213)

	at org.apache.flume.conf.FlumeConfiguration.validateConfiguration(FlumeConfiguration.java:127)

	at org.apache.flume.conf.FlumeConfiguration.<init>(FlumeConfiguration.java:109)

	at org.apache.flume.node.PropertiesFileConfigurationProvider.getFlumeConfiguration(PropertiesFileConfigurationProvider.java:189)

	at org.apache.flume.node.AbstractConfigurationProvider.getConfiguration(AbstractConfigurationProvider.java:89)

	at org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable.run(PollingPropertiesFileConfigurationProvider.java:140)

	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)

19/02/21 03:50:54 WARN conf.FlumeConfiguration: Could not configure sink  k2 due to: Channel k2 not in active set.

org.apache.flume.conf.ConfigurationException: Channel k2 not in active set.

	at org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.validateSinks(FlumeConfiguration.java:685)

	at org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.isValid(FlumeConfiguration.java:347)

	at org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.access$000(FlumeConfiguration.java:213)

	at org.apache.flume.conf.FlumeConfiguration.validateConfiguration(FlumeConfiguration.java:127)

	at org.apache.flume.conf.FlumeConfiguration.<init>(FlumeConfiguration.java:109)

	at org.apache.flume.node.PropertiesFileConfigurationProvider.getFlumeConfiguration(PropertiesFileConfigurationProvider.java:189)

	at org.apache.flume.node.AbstractConfigurationProvider.getConfiguration(AbstractConfigurationProvider.java:89)

	at org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable.run(PollingPropertiesFileConfigurationProvider.java:140)

	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)

# set channel for sinks

a1.sinks.k1.channel = c1

a1.sinks.k2.channel = k2

# set channel for sinks

a1.sinks.k1.channel = c1

a1.sinks.k2.channel = c2

# List the sources, sinks and channels for the agent

# <Agent>.sources = <Source1>

# <Agent>.sinks = <Sink1> <Sink2>

# <Agent>.channels = <Channel1> <Channel2>

#

# # set list of channels for source (separated by space)

# <Agent>.sources.<Source1>.channels = <Channel1> <Channel2>

#

# # set channel for sinks

# <Agent>.sinks.<Sink1>.channel = <Channel1>

# <Agent>.sinks.<Sink2>.channel = <Channel2>

#

# <Agent>.sources.<Source1>.selector.type = replicating

# Name the components on this agent

a1.sources = r1

a1.sinks = k1 k2

a1.channels = c1 c2

# Describe/configure the source

a1.sources.r1.type = exec

a1.sources.r1.command = tail -F /tmp/logs/cmcc.log

# Describe the sink

a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink

a1.sinks.k1.topic = cmcc1

a1.sinks.k1.brokerList = hdp1:9092,hdp2:9092,hdp3:9092

a1.sinks.k1.requiredAcks = 1

a1.sinks.k1.batchSize = 20

a1.sinks.k2.type = org.apache.flume.sink.kafka.KafkaSink

a1.sinks.k2.topic = cmcc2

a1.sinks.k2.brokerList = hdp1:9092,hdp2:9092,hdp3:9092

a1.sinks.k2.requiredAcks = 1

a1.sinks.k1.batchSize = 20

a1.sinks.k2.type = org.apache.flume.sink.kafka.KafkaSink

a1.sinks.k2.topic = cmcc2

a1.sinks.k2.brokerList = hdp1:9092,hdp2:9092,hdp3:9092

a1.sinks.k2.requiredAcks = 1

a1.sinks.k2.batchSize = 20

# Use a channel which buffers events in memory

a1.channels.c1.type = memory

a1.channels.c1.capacity = 1000

a1.channels.c1.transactionCapacity = 100

a1.channels.c2.type = file

a1.channels.c2.checkpointDir = /tmp/flume/checkpoint

a1.channels.c2.dataDirs = /tmp/flume/data

# Bind the source and sink to the channel

a1.sources.r1.channels = c1 c2

# set channel for sinks

a1.sinks.k1.channel = c1

a1.sinks.k2.channel = c2

 # #

a1.sources.r1.selector.type = replicating

 #