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

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
 #