KAFKA-6150: Purge repartition topics #4270

guozhangwang · 2017-11-28T18:00:36Z

Add the repartition topics information into ProcessorTopology: personally I do not like leaking this information into the topology but it seems not other simple way around.
StreamTask: added one more function to expose the consumed offsets from repartition topics only.
TaskManager: use the AdminClient to send the gathered offsets to delete only if a) previous call has completed and client intentionally ignore-and-log any errors, or b) no requests have ever called before.

NOTE that this code depends on the assumption that purge is only called right after the commit has succeeded, hence we presume all consumed offsets are committed.

MINOR: Added a few more constructor for ProcessorTopology for cleaner unit tests.
MINOR: Extracted MockStateStore out of the deprecated class.
MINOR: Made a pass over some unit test classes for clean ups.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

…into K6170-admin-client

…into K6150-purge-repartition-topics

guozhangwang · 2017-11-30T01:15:54Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java

@@ -890,17 +888,17 @@ public synchronized ProcessorTopology buildGlobalStateTopology() {
    }

    private ProcessorTopology build(final Set<String> nodeGroup) {
-        final List<ProcessorNode> processorNodes = new ArrayList<>(nodeFactories.size());


Removed the arraylist and use linked hashmap for processorMap to achieve the same thing: following insertion ordering.

…into K6150-purge-repartition-topics

guozhangwang · 2017-12-01T01:35:31Z

@dguy @mjsax @bbejeck this is ready to review.

dguy · 2017-12-01T13:56:34Z

@guozhangwang should this PR have all of those commits in it?

guozhangwang · 2017-12-01T16:37:36Z

@dguy yeah a lot of those are rebasing from trunk and other branches, and many others are minor commits.

guozhangwang · 2017-12-01T16:46:14Z

A few more notes on the PR itself:

Today no repartition topics are used as the changelog topics of state store, so re-processing / restoration is not affected by this change. But we need to keep in mind to exclude repartition topics used for changelogs if optimizations make them so.
I am still working on the integration test of it, but I'd rather have another PR after this to not make it too large.
There is another change I need to make: set the default segment size of repartition topic partitions to a small value (I'm thinking 50MB) so that record deletion can kicks in. But I may do that after KAFKA-4857: Replace StreamsKafkaClient with AdminClient in Kafka Streams #4242

dguy

Thanks @guozhangwang, left some comments.

dguy · 2017-12-01T13:58:15Z

clients/src/main/java/org/apache/kafka/clients/admin/DeleteRecordsResult.java

@@ -33,7 +33,7 @@

    private final Map<TopicPartition, KafkaFuture<DeletedRecords>> futures;

-    DeleteRecordsResult(Map<TopicPartition, KafkaFuture<DeletedRecords>> futures) {
+    public DeleteRecordsResult(Map<TopicPartition, KafkaFuture<DeletedRecords>> futures) {


making this public means it becomes part of the public API - right?

Good point. I will just keep the KafkaFuture from the result in TaskManager then.

@dguy Tried to remove this constructor from TaskManagerTest but it turns out ugly. So I plan to use the MockAdminClient after #4242 instead of the mock and create the object.

dguy · 2017-12-01T17:42:15Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorTopology.java

+                Collections.<String>emptySet());
+    }
+
+    public static ProcessorTopology withSources(final List<ProcessorNode> processorNodes,


package private?

dguy · 2017-12-01T17:42:25Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorTopology.java

+                Collections.<String>emptySet());
+    }
+
+    public static ProcessorTopology withLocalStores(final List<StateStore> stateStores,


package private ?

dguy · 2017-12-01T17:42:35Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorTopology.java

+                Collections.<String>emptySet());
+    }
+
+    public static ProcessorTopology withGlobalStores(final List<StateStore> stateStores,


dguy · 2017-12-01T17:42:48Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorTopology.java

+                Collections.<String>emptySet());
+    }
+
+    public static ProcessorTopology withRepartitionTopics(final List<ProcessorNode> processorNodes,


dguy · 2017-12-01T17:43:44Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorTopology.java

    }

    public Set<SinkNode> sinks() {
-        return new HashSet<>(sinkByTopics.values());
+        return new HashSet<>(sinksByTopic.values());


dguy · 2017-12-01T17:44:08Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorTopology.java

    public Map<String, String> storeToChangelogTopic() {
        return storeToChangelogTopic;
    }

-    public List<StateStore> globalStateStores() {
-        return globalStateStores;
+    public boolean isTopicInternalTransient(String topic) {


package private ?

maybe name it isInternalTransientTopic?

isInternalTopicTransient ?

dguy · 2017-12-01T17:46:15Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java

+        final Map<TopicPartition, Long> purgableConsumedOffsets = new HashMap<>();
+        for (Map.Entry<TopicPartition, Long> entry : consumedOffsets.entrySet()) {
+            if (topology.isTopicInternalTransient(entry.getKey().topic()))
+                purgableConsumedOffsets.put(entry.getKey(), entry.getValue() + 1);


is that right to purge the consumed offset + 1?

nit: extract .getKey() and call only once.

is that right to purge the consumed offset + 1?

The delete records call is to delete records before the given offset, similar to commit offset, so this is correct.

dguy · 2017-12-01T17:48:10Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java

+        if (deleteRecordsResult == null || deleteRecordsResult.all().isDone()) {
+
+            if (deleteRecordsResult != null && deleteRecordsResult.all().isCompletedExceptionally()) {
+                log.info("Previous delete-records request has failed: {}. Try sending the new request now", deleteRecordsResult.lowWatermarks());


does this need to be info level? seems more like debug to me, i.e., is a user actually going to care about this?

Do we want to print the whole map as-is? This might be quite large. Maybe filter for failed request?

dguy · 2017-12-01T17:52:22Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

@@ -542,6 +552,39 @@ public void shouldPropagateExceptionFromStandbyCommit() {
        verify(standby);
    }

+    @Test
+    public void shouldSendPurgeData() {


should we add a test that also checks that a purge request is sent after a previous one has completed?

mjsax

Thanks for the patch! Couple of comments.

mjsax · 2017-12-01T20:14:25Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java

        if (!factory.isWindowStore()) {
            return new InternalTopicConfig(name,
                                           Collections.singleton(InternalTopicConfig.CleanupPolicy.compact),
                                           factory.logConfig());
+        } else {
+            final InternalTopicConfig config = new InternalTopicConfig(name,


nit: move name to next line

mjsax · 2017-12-01T21:04:15Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java

        if (!factory.isWindowStore()) {
            return new InternalTopicConfig(name,
                                           Collections.singleton(InternalTopicConfig.CleanupPolicy.compact),
                                           factory.logConfig());
+        } else {


Nit: We don't need and else here as the if part end with return -- if you prefer if-else pattern, maybe more return after if-else block (seems to be cleaner to me -- but it's quite subjective)

mjsax · 2017-12-01T21:07:19Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorTopology.java

    public Map<String, String> storeToChangelogTopic() {
        return storeToChangelogTopic;
    }

-    public List<StateStore> globalStateStores() {
-        return globalStateStores;
+    public boolean isTopicInternalTransient(String topic) {


isInternalTopicTransient ?

mjsax · 2017-12-01T21:30:57Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java

@@ -364,6 +365,16 @@ private void commitOffsets(final boolean startNewTransaction) {
        }
    }

+    Map<TopicPartition, Long> purgableOffsets() {
+        final Map<TopicPartition, Long> purgableConsumedOffsets = new HashMap<>();
+        for (Map.Entry<TopicPartition, Long> entry : consumedOffsets.entrySet()) {


nit: add final

mjsax · 2017-12-01T21:31:35Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java

+        final Map<TopicPartition, Long> purgableConsumedOffsets = new HashMap<>();
+        for (Map.Entry<TopicPartition, Long> entry : consumedOffsets.entrySet()) {
+            if (topology.isTopicInternalTransient(entry.getKey().topic()))
+                purgableConsumedOffsets.put(entry.getKey(), entry.getValue() + 1);


nit: extract .getKey() and call only once.

mjsax · 2017-12-01T22:09:45Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java

+        assertThat(map, equalTo(Collections.singletonMap(repartition, 11L)));
+    }
+
+    private StreamTask createStatefulTask(boolean eosEnabled, boolean logged) {


nit: add final (same below)

mjsax · 2017-12-01T22:10:32Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java

@@ -328,7 +328,7 @@ public void shouldCommitAfterTheCommitInterval() {

    @SuppressWarnings({"ThrowableNotThrown", "unchecked"})
    private TaskManager mockTaskManagerCommit(final Consumer<byte[], byte[]> consumer, final int numberOfCommits, final int commits) {
-        final TaskManager taskManager = EasyMock.createMock(TaskManager.class);
+        final TaskManager taskManager = EasyMock.createNiceMock(TaskManager.class);


It has one more functions that is not focus of the test now.

mjsax · 2017-12-01T22:14:25Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

+
+        EasyMock.expect(active.recordsToDelete()).andReturn(recordsToDelete).once();
+        EasyMock.expect(adminClient.deleteRecords(recordsToDelete)).andReturn(deleteRecordsResult).once();
+        replay();


Do we not need to pass in active and adminClient here?

replay() will replay all the mock clients.

mjsax · 2017-12-01T22:16:53Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

+    public void shouldIgnorePurgeDataErrors() {
+        final KafkaFutureImpl<DeletedRecords> futureDeletedRecords = new KafkaFutureImpl<>();
+        final Map<TopicPartition, RecordsToDelete> recordsToDelete = Collections.singletonMap(t1p1, RecordsToDelete.beforeOffset(5L));
+        final DeleteRecordsResult deleteRecordsResult = new DeleteRecordsResult(Collections.singletonMap(t1p1, (KafkaFuture<DeletedRecords>) futureDeletedRecords));


Cast unnecessary?

Unfortunately it is not, since we are using KafkaFutureImpl for this test (but not the other) in order to call completeExceptionally.

Can't we specify the type via Collections.<KafkaFuture>singletonMap(..) to make it work? A Map of type KafkaFuture should accept KafkaFutureImpl -- or do I miss something? Generics can be confusion sometimes :)

mjsax · 2017-12-01T22:19:33Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

+
+        EasyMock.expect(active.recordsToDelete()).andReturn(recordsToDelete).times(2);
+        EasyMock.expect(adminClient.deleteRecords(recordsToDelete)).andReturn(deleteRecordsResult).times(2);
+        replay();


as above. pass in mocks?

guozhangwang · 2017-12-02T01:06:24Z

Addressed comments.

dguy · 2017-12-04T10:50:39Z

retest this please

dguy

Thanks @guozhangwang, LGTM

guozhangwang · 2017-12-04T18:22:16Z

Thanks for the reviews! Merged to trunk.

guozhangwang added 30 commits November 6, 2017 16:14

add AdminClient

6c5b20e

Merge branch 'trunk' of https://git-wip-us.apache.org/repos/asf/kafka …

fc908e0

…into K6170-admin-client

refactor thread / assignor dependency

d1be566

revert TaskManager APIs

d1a778f

move logic of assignor to task manager

f9e5fbf

minor fix

bfd08c4

extract delegating restore listener

f95dc0b

add admin configs in streams config

10ceff0

add AdminClient to stream thread

41dc2b0

Merge branch 'trunk' of https://git-wip-us.apache.org/repos/asf/kafka …

3592206

…into K6170-admin-client

add admin prefix

03e64d0

merge from K6170-admin-client

035b3a6

minor fixes

40cb06a

Merge branch 'trunk' of https://git-wip-us.apache.org/repos/asf/kafka …

682cb4b

…into K6170-admin-client

github comments

4502fa6

github comments

b2a34a2

merged from K6170-admin-client

64ae49d

remove unused imports

853436e

fix unit tests

0643709

more unit tests

7fc5a0d

remove unused imports

44986d8

merge from apache trunk

0d3a4c8

github comments

5735d6d

refactor StreamPartitionAssignorTest

f7a4ba2

one critical bug fix

d9c0e99

fix unit tests

e8c43c3

minor fix

5358356

make originalReset from consumer configs

a18b4d5

expose committed offsets

3e3938d

github comments

14e16ab

guozhangwang added 7 commits November 28, 2017 09:59

merge from trunk

1d13404

remove partitionsByHostState

14ce324

revert changes for obeying state store insertion ordering

1570eb7

remove unused imports

305aa02

refactor StreamTaskTest

3a218ca

minor change

88ce7eb

add unit tests

e645dfe

guozhangwang changed the title ~~[WIP] KAFKA-6150: Purge repartition topics~~ KAFKA-6150: Purge repartition topics Nov 29, 2017

guozhangwang added 4 commits November 29, 2017 11:04

Merge branch 'trunk' of https://git-wip-us.apache.org/repos/asf/kafka …

856c983

…into K6150-purge-repartition-topics

add missing MockStateStore.java

1f42605

remove unused imports

6069ce2

minor fixes

9ba2ec3

guozhangwang commented Nov 30, 2017

View reviewed changes

guozhangwang added 2 commits November 30, 2017 17:01

Merge branch 'trunk' of https://git-wip-us.apache.org/repos/asf/kafka …

bc4f7df

…into K6150-purge-repartition-topics

one minor unit test fix

d032923

dguy reviewed Dec 1, 2017

View reviewed changes

mjsax reviewed Dec 1, 2017

View reviewed changes

github comments

0f203a9

minor fix

c73bc8d

dguy approved these changes Dec 4, 2017

View reviewed changes

asfgit closed this in 4b8a29f Dec 4, 2017

guozhangwang deleted the K6150-purge-repartition-topics branch February 14, 2018 19:51

KAFKA-6150: Purge repartition topics #4270

KAFKA-6150: Purge repartition topics #4270

Conversation

guozhangwang commented Nov 28, 2017 • edited Loading

Committer Checklist (excluded from commit message)

Choose a reason for hiding this comment

guozhangwang commented Dec 1, 2017

dguy commented Dec 1, 2017

guozhangwang commented Dec 1, 2017

guozhangwang commented Dec 1, 2017

dguy left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang commented Dec 2, 2017

dguy commented Dec 4, 2017

dguy left a comment

Choose a reason for hiding this comment

guozhangwang commented Dec 4, 2017

guozhangwang commented Nov 28, 2017 •

edited

Loading