False [Topic authorization failed] error for topic

[dgyimesi]

Describe the bug

Broker reports [Topic authorization failed] for a topic that KafkaJS could publish to before in the same Producer instance.
Broker provider: Confluent

To Reproduce

1. set up ACL for your API key (eg. this.is.my.env, PREFIXED)
2. create a producer instance
3. send a message to a topic that matches your ACL, like 'this.is.my.env.created', it will pass
4. send a message to a topic that does NOT match your ACL, like 'this.is.my.otherenv.created', it will fail with error above (as expected)
5. as in step 3. send a message to a topic that matches your ACL, like 'this.is.my.env.created', it will fail again with the same error (unexpected)

Expected behavior
Messages that matches the ACL will pass even if there were failing attempts before

Environment:

• OS: [e.g. Windows 10]
• KafkaJS version [e.g. 1.14.0, 1.15.0, 2.0.0]
• NodeJS version [e.g. 16.14.0]

Additional context
During debugging I found that the Metadata protocol message buffer contains the forbidden topic name after encoding, even at step 5. send() gets only one topic (that matches the ACL) as a parameter, still the buffer contains the topic name from my previous send attempt.

[Nevon]

I was able to reproduce this here:

kafkajs/src/admin/__tests__/createAcls.spec.js

Lines 255 to 304 in 8cbfb1b

	test('Repro 1346: Producing to allowed topic after failing to produce to not-allowed topic', async () => {
	const allowedTopic = `allowed-${secureRandom()}`
	const notAllowedTopic = `disallowed-${secureRandom()}`
	admin = createSASLAdminClientForUser({ username: 'test', password: 'testtest' })
	await admin.connect()
	await admin.createTopics({
	waitForLeaders: true,
	topics: [allowedTopic, notAllowedTopic].map(topic => ({ topic, numPartitions: 1 })),
	})
	await admin.createAcls({
	acl: [
	{
	resourceType: ACL_RESOURCE_TYPES.TOPIC,
	resourceName: notAllowedTopic,
	resourcePatternType: RESOURCE_PATTERN_TYPES.LITERAL,
	principal: 'User:bob',
	host: '*',
	operation: ACL_OPERATION_TYPES.WRITE,
	permissionType: ACL_PERMISSION_TYPES.DENY,
	},
	{
	resourceType: ACL_RESOURCE_TYPES.TOPIC,
	resourceName: allowedTopic,
	resourcePatternType: RESOURCE_PATTERN_TYPES.LITERAL,
	principal: 'User:bob',
	host: '*',
	operation: ACL_OPERATION_TYPES.WRITE,
	permissionType: ACL_PERMISSION_TYPES.ALLOW,
	},
	],
	})
	await admin.disconnect()
	const producer = createSASLProducerClientForUser({ username: 'bob', password: 'bobbob' })
	await producer.connect()
	await expect(
	producer.send({ topic: allowedTopic, messages: [{ value: 'hello' }] })
	).resolves.not.toBeUndefined()
	await expect(
	producer.send({ topic: notAllowedTopic, messages: [{ value: 'whoops' }] })
	).rejects.not.toBeUndefined()
	await expect(
	producer.send({ topic: allowedTopic, messages: [{ value: 'world' }] })
	).resolves.not.toBeUndefined()
	await producer.disconnect()
	})

Sharing the results of my debugging: What's happening is that the topic gets added to the cluster's list of "target topics", which is the list of topics it'll fetch metadata for when needed. So on the first produce, the allowed topic would be added to the list, and everything is fine. On the second produce, the not allowed topic is added to the list and we get an error back. On the third produce, we still try to fetch metadata for the not allowed topic, because it's still in the list of target topics, so we get an error.

There are two solutions:

1. Remove the topic from the target topic list when getting a topic authorization error. This requires surfacing this error better from the metadata error, as currently we lose the information about which topic(s) caused the error.
2. Remove the target topics construct entirely, since we don't really need it. There was an attempt to do this in Eliminate target topics list on the cluster #667, which I think is the right way in the long term.

[dgyimesi]

Since my report I found an additional use case which leads to the same issue. Regardless of ACLs on your topics publishing to an existing topic after an attempt to publish to a non-existing one has the same effect, except that the error message will be "This server does not host this topic-partition", presumably because of the very same beheviour on metadata fetching as described above.

[dgyimesi]

@Nevon as I see #667 is still open and not approved. Any plans to incorporate it in the next release?

[mattia-crypto]

I have a similar issue here: #1377

[Nevon]

@Nevon as I see #667 is still open and not approved. Any plans to incorporate it in the next release?

That PR requires quite a lot of refactoring, since we've restructured a huge chunk of the consumer since that one was last touched. I spent my afternoon trying to bring it up to date, and I plan on going through it again properly tomorrow. I'd like to get it merged because it will make #1040 much more doable, as well as solve this issue and #1185.

[Nevon]

Sorry to disappoint, but the direction in #667 isn't going to work, as I realized once I spent a few days on it and the cracks started to show. I've elaborated in #667 (comment)

In the meantime, I think this specific issue can probably be solved with a bit of a quick-fix by just removing the topic from targetTopics when we get an authorization error on metadata refresh. There might be some issues with that as well, specifically for consumers where you always want targetTopics to reflect the topics you're subscribed to, so maybe it's a producer-specific behavior.

[dgyimesi]

@Nevon no worries, thank you for letting me know. I will prepare a PR with the recommended fix.

[Nevon]

Fixed by #1385