From 40227737cc2c25020e6ab9313c0eb6e82c095d54 Mon Sep 17 00:00:00 2001 From: soukhand Date: Thu, 4 Jul 2024 15:30:10 +0530 Subject: [PATCH] commiting including unsupported test files --- LICENSE.txt | 2 +- build.gradle | 2 +- .../clients/ClusterConnectionStates.java | 2 +- .../okafka/clients/CommonClientConfigs.java | 2 +- .../oracle/okafka/clients/KafkaClient.java | 2 +- .../org/oracle/okafka/clients/Metadata.java | 2 +- .../oracle/okafka/clients/NetworkClient.java | 2 +- .../okafka/clients/TopicTeqParameters.java | 7 + .../oracle/okafka/clients/admin/Admin.java | 7 + .../okafka/clients/admin/AdminClient.java | 2 +- .../clients/admin/AdminClientConfig.java | 2 +- .../clients/admin/CreateTopicsResult.java | 7 + .../clients/admin/DeleteTopicsOptions.java | 7 + .../clients/admin/DeleteTopicsResult.java | 7 + .../clients/admin/KafkaAdminClient.java | 2 +- .../clients/admin/internals/AQKafkaAdmin.java | 2 +- .../clients/consumer/ConsumerConfig.java | 2 +- .../clients/consumer/KafkaConsumer.java | 2 +- .../okafka/clients/consumer/TxEQAssignor.java | 7 + .../consumer/internals/AQKafkaConsumer.java | 2 +- .../internals/ConsumerNetworkClient.java | 2 +- .../consumer/internals/SubscriptionState.java | 7 + .../clients/producer/KafkaProducer.java | 7 + .../clients/producer/ProducerConfig.java | 2 +- .../producer/internals/AQKafkaProducer.java | 2 +- .../internals/FutureRecordMetadata.java | 2 +- .../producer/internals/IncompleteBatches.java | 7 + .../internals/OracleTransactionManager.java | 7 + .../internals/ProduceRequestResult.java | 2 +- .../producer/internals/ProducerBatch.java | 7 + .../internals/ProducerIdAndEpoch.java | 2 +- .../producer/internals/RecordAccumulator.java | 7 + .../internals/SenderMetricsRegistry.java | 7 + .../producer/internals/SenderThread.java | 2 +- .../org/oracle/okafka/common/AQException.java | 2 +- .../java/org/oracle/okafka/common/Node.java | 2 +- .../okafka/common/config/SslConfigs.java | 2 +- .../common/errors/ConnectionException.java | 2 +- .../errors/FeatureNotSupportedException.java | 2 +- .../InvalidLoginCredentialsException.java | 2 +- .../errors/InvalidMessageIdException.java | 2 +- .../okafka/common/network/AQClient.java | 2 +- .../okafka/common/protocol/ApiKeys.java | 2 +- .../okafka/common/record/BaseRecords.java | 2 +- .../okafka/common/record/BufferSupplier.java | 2 +- .../common/requests/AbstractRequest.java | 2 +- .../common/requests/AbstractResponse.java | 2 +- .../okafka/common/requests/CommitRequest.java | 2 +- .../common/requests/CommitResponse.java | 2 +- .../common/requests/CreateTopicsRequest.java | 2 +- .../common/requests/CreateTopicsResponse.java | 2 +- .../common/requests/DeleteTopicsRequest.java | 2 +- .../common/requests/DeleteTopicsResponse.java | 2 +- .../okafka/common/requests/FetchRequest.java | 2 +- .../okafka/common/requests/FetchResponse.java | 2 +- .../common/requests/IsolationLevel.java | 2 +- .../common/requests/MetadataRequest.java | 2 +- .../common/requests/MetadataResponse.java | 2 +- .../common/requests/OffsetResetRequest.java | 2 +- .../common/requests/OffsetResetResponse.java | 2 +- .../common/requests/ProduceRequest.java | 2 +- .../common/requests/ProduceResponse.java | 2 +- .../okafka/common/requests/RequestHeader.java | 2 +- .../common/requests/ResponseHeader.java | 2 +- .../common/requests/SubscribeRequest.java | 2 +- .../common/requests/SubscribeResponse.java | 2 +- .../common/requests/UnsubscribeRequest.java | 2 +- .../common/requests/UnsubscribeResponse.java | 2 +- .../okafka/common/utils/ConnectionUtils.java | 2 +- .../okafka/common/utils/CreateTopics.java | 2 +- .../common/utils/MessageIdConverter.java | 2 +- .../oracle/okafka/common/utils/TNSParser.java | 2 +- .../org/oracle/okafka/clients/MockClient.java | 18 ++- .../clients/consumer/ConsumerConfigTest.java | 2 +- .../clients/consumer/ConsumerRecordTest.java | 2 +- .../clients/consumer/ConsumerRecordsTest.java | 2 +- ...izeCompatibilityOffsetAndMetadataTest.java | 2 +- .../clients/producer/KafkaProducerTest.java | 53 +++++-- .../clients/producer/ProducerRecordTest.java | 4 +- .../clients/producer/RecordMetadataTest.java | 7 +- .../clients/producer/RecordSendTest.java | 15 +- .../producer/internals/BufferPoolTest.java | 33 +++-- .../internals/DefaultPartitionerTest.java | 9 +- .../producer/internals/ProducerBatchTest.java | 34 ++--- .../internals/ProducerInterceptorsTest.java | 13 +- .../internals/RecordAccumulatorTest.java | 40 ++--- .../oracle/okafka/common/utils/MockTime.java | 46 ++++-- .../okafka/common/utils/Serializer.java | 2 +- .../okafka/test/MockMetricsReporter.java | 6 +- .../oracle/okafka/test/MockPartitioner.java | 6 +- .../okafka/test/MockProducerInterceptor.java | 14 +- .../oracle/okafka/test/MockSerializer.java | 8 +- .../org/oracle/okafka/test/TestCondition.java | 2 +- .../org/oracle/okafka/test/TestUtils.java | 8 +- .../okafka/tests/ConsumerMetricsTest.java | 137 ++++++++++++++++++ .../okafka/tests/OkafkaAutoOffsetReset.java | 76 ++++++++++ .../okafka/tests/OkafkaDeleteTopic.java | 39 +++++ .../okafka/tests/OkafkaSeekToBeginning.java | 97 +++++++++++++ .../oracle/okafka/tests/OkafkaSeekToEnd.java | 99 +++++++++++++ .../org/oracle/okafka/tests/OkafkaSetup.java | 26 ++++ .../okafka/tests/OkafkaUnsubscribe.java | 106 ++++++++++++++ .../okafka/tests/ProducerMetricsTest.java | 112 ++++++++++++++ .../okafka/tests/SimpleOkafkaAdmin.java | 41 ++++++ .../okafka/tests/SimpleOkafkaConsumer.java | 69 +++++++++ .../okafka/tests/SimpleOkafkaProducer.java | 42 ++++++ .../org/oracle/okafka/tests/TestRunner.java | 26 ++++ .../org/oracle/okafka/examples/Consumer.java | 2 +- .../org/oracle/okafka/examples/Producer.java | 2 +- settings.gradle | 2 +- 109 files changed, 1214 insertions(+), 197 deletions(-) create mode 100644 clients/src/test/java/org/oracle/okafka/tests/ConsumerMetricsTest.java create mode 100644 clients/src/test/java/org/oracle/okafka/tests/OkafkaAutoOffsetReset.java create mode 100644 clients/src/test/java/org/oracle/okafka/tests/OkafkaDeleteTopic.java create mode 100644 clients/src/test/java/org/oracle/okafka/tests/OkafkaSeekToBeginning.java create mode 100644 clients/src/test/java/org/oracle/okafka/tests/OkafkaSeekToEnd.java create mode 100644 clients/src/test/java/org/oracle/okafka/tests/OkafkaSetup.java create mode 100644 clients/src/test/java/org/oracle/okafka/tests/OkafkaUnsubscribe.java create mode 100644 clients/src/test/java/org/oracle/okafka/tests/ProducerMetricsTest.java create mode 100644 clients/src/test/java/org/oracle/okafka/tests/SimpleOkafkaAdmin.java create mode 100644 clients/src/test/java/org/oracle/okafka/tests/SimpleOkafkaConsumer.java create mode 100644 clients/src/test/java/org/oracle/okafka/tests/SimpleOkafkaProducer.java create mode 100644 clients/src/test/java/org/oracle/okafka/tests/TestRunner.java diff --git a/LICENSE.txt b/LICENSE.txt index a93828e..0fce3de 100644 --- a/LICENSE.txt +++ b/LICENSE.txt @@ -1,4 +1,4 @@ -Copyright (c) 2019, 2023 Oracle and/or its affiliates. +Copyright (c) 2019, 2024 Oracle and/or its affiliates. The Universal Permissive License (UPL), Version 1.0 diff --git a/build.gradle b/build.gradle index ef35216..3be4a0e 100644 --- a/build.gradle +++ b/build.gradle @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2022 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/clients/ClusterConnectionStates.java b/clients/src/main/java/org/oracle/okafka/clients/ClusterConnectionStates.java index bcb0a82..eb6ff9d 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/ClusterConnectionStates.java +++ b/clients/src/main/java/org/oracle/okafka/clients/ClusterConnectionStates.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/clients/CommonClientConfigs.java b/clients/src/main/java/org/oracle/okafka/clients/CommonClientConfigs.java index df23647..8bf80b9 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/CommonClientConfigs.java +++ b/clients/src/main/java/org/oracle/okafka/clients/CommonClientConfigs.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/clients/KafkaClient.java b/clients/src/main/java/org/oracle/okafka/clients/KafkaClient.java index d73d0a4..4816daa 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/KafkaClient.java +++ b/clients/src/main/java/org/oracle/okafka/clients/KafkaClient.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/clients/Metadata.java b/clients/src/main/java/org/oracle/okafka/clients/Metadata.java index 95149c7..9f27fd4 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/Metadata.java +++ b/clients/src/main/java/org/oracle/okafka/clients/Metadata.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** - ** Copyright (c) 2019, 2020 Oracle and/or its affiliates. + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/clients/NetworkClient.java b/clients/src/main/java/org/oracle/okafka/clients/NetworkClient.java index 54961fb..ad2d414 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/NetworkClient.java +++ b/clients/src/main/java/org/oracle/okafka/clients/NetworkClient.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** - ** Copyright (c) 2019, 2020 Oracle and/or its affiliates. + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/clients/TopicTeqParameters.java b/clients/src/main/java/org/oracle/okafka/clients/TopicTeqParameters.java index e03a4f8..dee90bd 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/TopicTeqParameters.java +++ b/clients/src/main/java/org/oracle/okafka/clients/TopicTeqParameters.java @@ -1,3 +1,10 @@ +/* +** OKafka Java Client version 23.4. +** +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. +** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. +*/ + package org.oracle.okafka.clients; diff --git a/clients/src/main/java/org/oracle/okafka/clients/admin/Admin.java b/clients/src/main/java/org/oracle/okafka/clients/admin/Admin.java index 2ff5be0..21a6c12 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/admin/Admin.java +++ b/clients/src/main/java/org/oracle/okafka/clients/admin/Admin.java @@ -1,3 +1,10 @@ +/* + ** OKafka Java Client version 23.4. + ** + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. + ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. + */ + /* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with diff --git a/clients/src/main/java/org/oracle/okafka/clients/admin/AdminClient.java b/clients/src/main/java/org/oracle/okafka/clients/admin/AdminClient.java index b678377..1a4f69a 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/admin/AdminClient.java +++ b/clients/src/main/java/org/oracle/okafka/clients/admin/AdminClient.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/clients/admin/AdminClientConfig.java b/clients/src/main/java/org/oracle/okafka/clients/admin/AdminClientConfig.java index 3693416..ae3660b 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/admin/AdminClientConfig.java +++ b/clients/src/main/java/org/oracle/okafka/clients/admin/AdminClientConfig.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/clients/admin/CreateTopicsResult.java b/clients/src/main/java/org/oracle/okafka/clients/admin/CreateTopicsResult.java index 47c53e8..afe67f9 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/admin/CreateTopicsResult.java +++ b/clients/src/main/java/org/oracle/okafka/clients/admin/CreateTopicsResult.java @@ -1,3 +1,10 @@ +/* + ** OKafka Java Client version 23.4. + ** + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. + ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. + */ + /* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with diff --git a/clients/src/main/java/org/oracle/okafka/clients/admin/DeleteTopicsOptions.java b/clients/src/main/java/org/oracle/okafka/clients/admin/DeleteTopicsOptions.java index 2dfdc92..97f4549 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/admin/DeleteTopicsOptions.java +++ b/clients/src/main/java/org/oracle/okafka/clients/admin/DeleteTopicsOptions.java @@ -1,3 +1,10 @@ +/* + ** OKafka Java Client version 23.4. + ** + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. + ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. + */ + /* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with diff --git a/clients/src/main/java/org/oracle/okafka/clients/admin/DeleteTopicsResult.java b/clients/src/main/java/org/oracle/okafka/clients/admin/DeleteTopicsResult.java index 6f9de9f..7d591e4 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/admin/DeleteTopicsResult.java +++ b/clients/src/main/java/org/oracle/okafka/clients/admin/DeleteTopicsResult.java @@ -1,3 +1,10 @@ +/* + ** OKafka Java Client version 23.4. + ** + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. + ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. + */ + /* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with diff --git a/clients/src/main/java/org/oracle/okafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/oracle/okafka/clients/admin/KafkaAdminClient.java index bc98d30..f46f967 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/admin/KafkaAdminClient.java +++ b/clients/src/main/java/org/oracle/okafka/clients/admin/KafkaAdminClient.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** - ** Copyright (c) 2019, 2020 Oracle and/or its affiliates. + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/clients/admin/internals/AQKafkaAdmin.java b/clients/src/main/java/org/oracle/okafka/clients/admin/internals/AQKafkaAdmin.java index f644400..357bb38 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/admin/internals/AQKafkaAdmin.java +++ b/clients/src/main/java/org/oracle/okafka/clients/admin/internals/AQKafkaAdmin.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/clients/consumer/ConsumerConfig.java b/clients/src/main/java/org/oracle/okafka/clients/consumer/ConsumerConfig.java index 7627c93..2b0c117 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/consumer/ConsumerConfig.java +++ b/clients/src/main/java/org/oracle/okafka/clients/consumer/ConsumerConfig.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/oracle/okafka/clients/consumer/KafkaConsumer.java index d5e1740..ca4ff32 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/consumer/KafkaConsumer.java +++ b/clients/src/main/java/org/oracle/okafka/clients/consumer/KafkaConsumer.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** - ** Copyright (c) 2019, 2020 Oracle and/or its affiliates. + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/clients/consumer/TxEQAssignor.java b/clients/src/main/java/org/oracle/okafka/clients/consumer/TxEQAssignor.java index 5403676..257e216 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/consumer/TxEQAssignor.java +++ b/clients/src/main/java/org/oracle/okafka/clients/consumer/TxEQAssignor.java @@ -1,3 +1,10 @@ +/* +** OKafka Java Client version 23.4. +** +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. +** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. +*/ + package org.oracle.okafka.clients.consumer; import java.util.ArrayList; diff --git a/clients/src/main/java/org/oracle/okafka/clients/consumer/internals/AQKafkaConsumer.java b/clients/src/main/java/org/oracle/okafka/clients/consumer/internals/AQKafkaConsumer.java index b0b09d9..307c14f 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/consumer/internals/AQKafkaConsumer.java +++ b/clients/src/main/java/org/oracle/okafka/clients/consumer/internals/AQKafkaConsumer.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/clients/consumer/internals/ConsumerNetworkClient.java b/clients/src/main/java/org/oracle/okafka/clients/consumer/internals/ConsumerNetworkClient.java index 92f82a3..12ba54a 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/consumer/internals/ConsumerNetworkClient.java +++ b/clients/src/main/java/org/oracle/okafka/clients/consumer/internals/ConsumerNetworkClient.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** - ** Copyright (c) 2019, 2020 Oracle and/or its affiliates. + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/clients/consumer/internals/SubscriptionState.java b/clients/src/main/java/org/oracle/okafka/clients/consumer/internals/SubscriptionState.java index 2793810..ecef791 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/consumer/internals/SubscriptionState.java +++ b/clients/src/main/java/org/oracle/okafka/clients/consumer/internals/SubscriptionState.java @@ -1,3 +1,10 @@ +/* + ** OKafka Java Client version 23.4. + ** + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. + ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. + */ + /* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with diff --git a/clients/src/main/java/org/oracle/okafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/oracle/okafka/clients/producer/KafkaProducer.java index 7f0101d..f7405f4 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/producer/KafkaProducer.java +++ b/clients/src/main/java/org/oracle/okafka/clients/producer/KafkaProducer.java @@ -1,3 +1,10 @@ +/* + ** OKafka Java Client version 23.4. + ** + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. + ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. + */ + /* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with diff --git a/clients/src/main/java/org/oracle/okafka/clients/producer/ProducerConfig.java b/clients/src/main/java/org/oracle/okafka/clients/producer/ProducerConfig.java index 581bd5f..765eff0 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/producer/ProducerConfig.java +++ b/clients/src/main/java/org/oracle/okafka/clients/producer/ProducerConfig.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** - ** Copyright (c) 2019, 2020 Oracle and/or its affiliates. + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/AQKafkaProducer.java b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/AQKafkaProducer.java index b94feab..4b91d47 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/AQKafkaProducer.java +++ b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/AQKafkaProducer.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** - ** Copyright (c) 2019, 2020 Oracle and/or its affiliates. + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/FutureRecordMetadata.java b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/FutureRecordMetadata.java index 4f9de46..8b9847d 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/FutureRecordMetadata.java +++ b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/FutureRecordMetadata.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/IncompleteBatches.java b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/IncompleteBatches.java index b365214..44e8278 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/IncompleteBatches.java +++ b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/IncompleteBatches.java @@ -1,3 +1,10 @@ +/* + ** OKafka Java Client version 23.4. + ** + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. + ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. + */ + /* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with diff --git a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/OracleTransactionManager.java b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/OracleTransactionManager.java index 1a584b0..23bc7d7 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/OracleTransactionManager.java +++ b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/OracleTransactionManager.java @@ -1,3 +1,10 @@ +/* +** OKafka Java Client version 23.4. +** +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. +** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. +*/ + package org.oracle.okafka.clients.producer.internals; import java.sql.Connection; diff --git a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/ProduceRequestResult.java b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/ProduceRequestResult.java index 044571b..863a57c 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/ProduceRequestResult.java +++ b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/ProduceRequestResult.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/ProducerBatch.java b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/ProducerBatch.java index b3d787b..2d093a6 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/ProducerBatch.java +++ b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/ProducerBatch.java @@ -1,3 +1,10 @@ +/* + ** OKafka Java Client version 23.4. + ** + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. + ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. + */ + /* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with diff --git a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/ProducerIdAndEpoch.java b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/ProducerIdAndEpoch.java index a99294f..13c9dfa 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/ProducerIdAndEpoch.java +++ b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/ProducerIdAndEpoch.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/RecordAccumulator.java b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/RecordAccumulator.java index 408d03c..64d20ef 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/RecordAccumulator.java +++ b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/RecordAccumulator.java @@ -1,3 +1,10 @@ +/* + ** OKafka Java Client version 23.4. + ** + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. + ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. + */ + /* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with diff --git a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/SenderMetricsRegistry.java b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/SenderMetricsRegistry.java index ce209f6..5fa68e7 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/SenderMetricsRegistry.java +++ b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/SenderMetricsRegistry.java @@ -1,3 +1,10 @@ +/* +** OKafka Java Client version 23.4. +** +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. +** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. +*/ + package org.oracle.okafka.clients.producer.internals; import java.util.ArrayList; diff --git a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/SenderThread.java b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/SenderThread.java index 4e5a4c9..01954f1 100644 --- a/clients/src/main/java/org/oracle/okafka/clients/producer/internals/SenderThread.java +++ b/clients/src/main/java/org/oracle/okafka/clients/producer/internals/SenderThread.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** - ** Copyright (c) 2019, 2020 Oracle and/or its affiliates. + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/AQException.java b/clients/src/main/java/org/oracle/okafka/common/AQException.java index 09c7ac5..abd583d 100644 --- a/clients/src/main/java/org/oracle/okafka/common/AQException.java +++ b/clients/src/main/java/org/oracle/okafka/common/AQException.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/Node.java b/clients/src/main/java/org/oracle/okafka/common/Node.java index de1b0b2..43ee9f4 100644 --- a/clients/src/main/java/org/oracle/okafka/common/Node.java +++ b/clients/src/main/java/org/oracle/okafka/common/Node.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/config/SslConfigs.java b/clients/src/main/java/org/oracle/okafka/common/config/SslConfigs.java index caef25e..08593cf 100644 --- a/clients/src/main/java/org/oracle/okafka/common/config/SslConfigs.java +++ b/clients/src/main/java/org/oracle/okafka/common/config/SslConfigs.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/errors/ConnectionException.java b/clients/src/main/java/org/oracle/okafka/common/errors/ConnectionException.java index 406024d..6c36f2a 100644 --- a/clients/src/main/java/org/oracle/okafka/common/errors/ConnectionException.java +++ b/clients/src/main/java/org/oracle/okafka/common/errors/ConnectionException.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/errors/FeatureNotSupportedException.java b/clients/src/main/java/org/oracle/okafka/common/errors/FeatureNotSupportedException.java index 047eba4..496b3d2 100644 --- a/clients/src/main/java/org/oracle/okafka/common/errors/FeatureNotSupportedException.java +++ b/clients/src/main/java/org/oracle/okafka/common/errors/FeatureNotSupportedException.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/errors/InvalidLoginCredentialsException.java b/clients/src/main/java/org/oracle/okafka/common/errors/InvalidLoginCredentialsException.java index 7f270ac..999267a 100644 --- a/clients/src/main/java/org/oracle/okafka/common/errors/InvalidLoginCredentialsException.java +++ b/clients/src/main/java/org/oracle/okafka/common/errors/InvalidLoginCredentialsException.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/errors/InvalidMessageIdException.java b/clients/src/main/java/org/oracle/okafka/common/errors/InvalidMessageIdException.java index f8136f7..74123c5 100644 --- a/clients/src/main/java/org/oracle/okafka/common/errors/InvalidMessageIdException.java +++ b/clients/src/main/java/org/oracle/okafka/common/errors/InvalidMessageIdException.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/network/AQClient.java b/clients/src/main/java/org/oracle/okafka/common/network/AQClient.java index b9fe63c..5f09d58 100644 --- a/clients/src/main/java/org/oracle/okafka/common/network/AQClient.java +++ b/clients/src/main/java/org/oracle/okafka/common/network/AQClient.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/protocol/ApiKeys.java b/clients/src/main/java/org/oracle/okafka/common/protocol/ApiKeys.java index a801204..76ba467 100644 --- a/clients/src/main/java/org/oracle/okafka/common/protocol/ApiKeys.java +++ b/clients/src/main/java/org/oracle/okafka/common/protocol/ApiKeys.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/record/BaseRecords.java b/clients/src/main/java/org/oracle/okafka/common/record/BaseRecords.java index 6a27379..8871f5c 100644 --- a/clients/src/main/java/org/oracle/okafka/common/record/BaseRecords.java +++ b/clients/src/main/java/org/oracle/okafka/common/record/BaseRecords.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/record/BufferSupplier.java b/clients/src/main/java/org/oracle/okafka/common/record/BufferSupplier.java index 03cd0b1..093fb3d 100644 --- a/clients/src/main/java/org/oracle/okafka/common/record/BufferSupplier.java +++ b/clients/src/main/java/org/oracle/okafka/common/record/BufferSupplier.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/AbstractRequest.java b/clients/src/main/java/org/oracle/okafka/common/requests/AbstractRequest.java index 57c6163..a086cb7 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/AbstractRequest.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/AbstractRequest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/AbstractResponse.java b/clients/src/main/java/org/oracle/okafka/common/requests/AbstractResponse.java index f0b9f4a..0960c07 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/AbstractResponse.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/AbstractResponse.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/CommitRequest.java b/clients/src/main/java/org/oracle/okafka/common/requests/CommitRequest.java index d899e4c..0352078 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/CommitRequest.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/CommitRequest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/CommitResponse.java b/clients/src/main/java/org/oracle/okafka/common/requests/CommitResponse.java index c3b1657..a685f48 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/CommitResponse.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/CommitResponse.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/CreateTopicsRequest.java b/clients/src/main/java/org/oracle/okafka/common/requests/CreateTopicsRequest.java index 74a71be..c0d4427 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/CreateTopicsRequest.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/CreateTopicsRequest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/CreateTopicsResponse.java b/clients/src/main/java/org/oracle/okafka/common/requests/CreateTopicsResponse.java index d76414d..bd3408f 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/CreateTopicsResponse.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/CreateTopicsResponse.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/DeleteTopicsRequest.java b/clients/src/main/java/org/oracle/okafka/common/requests/DeleteTopicsRequest.java index cb0d81e..2342b56 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/DeleteTopicsRequest.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/DeleteTopicsRequest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/DeleteTopicsResponse.java b/clients/src/main/java/org/oracle/okafka/common/requests/DeleteTopicsResponse.java index f4ae87d..1719921 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/DeleteTopicsResponse.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/DeleteTopicsResponse.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/FetchRequest.java b/clients/src/main/java/org/oracle/okafka/common/requests/FetchRequest.java index ecafe58..7a558a2 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/FetchRequest.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/FetchRequest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/FetchResponse.java b/clients/src/main/java/org/oracle/okafka/common/requests/FetchResponse.java index 95651c8..fbec6f7 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/FetchResponse.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/FetchResponse.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/IsolationLevel.java b/clients/src/main/java/org/oracle/okafka/common/requests/IsolationLevel.java index 47c1c18..82be32a 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/IsolationLevel.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/IsolationLevel.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/MetadataRequest.java b/clients/src/main/java/org/oracle/okafka/common/requests/MetadataRequest.java index 162361e..14599e5 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/MetadataRequest.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/MetadataRequest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/MetadataResponse.java b/clients/src/main/java/org/oracle/okafka/common/requests/MetadataResponse.java index a71dd2d..cf013d0 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/MetadataResponse.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/MetadataResponse.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/OffsetResetRequest.java b/clients/src/main/java/org/oracle/okafka/common/requests/OffsetResetRequest.java index dfce500..9db017e 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/OffsetResetRequest.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/OffsetResetRequest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/OffsetResetResponse.java b/clients/src/main/java/org/oracle/okafka/common/requests/OffsetResetResponse.java index 2a0d80e..c123697 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/OffsetResetResponse.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/OffsetResetResponse.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/ProduceRequest.java b/clients/src/main/java/org/oracle/okafka/common/requests/ProduceRequest.java index f105b45..839dfb9 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/ProduceRequest.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/ProduceRequest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/ProduceResponse.java b/clients/src/main/java/org/oracle/okafka/common/requests/ProduceResponse.java index 8c58c0f..92ac1b6 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/ProduceResponse.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/ProduceResponse.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/RequestHeader.java b/clients/src/main/java/org/oracle/okafka/common/requests/RequestHeader.java index d5f2637..b22941a 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/RequestHeader.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/RequestHeader.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/ResponseHeader.java b/clients/src/main/java/org/oracle/okafka/common/requests/ResponseHeader.java index 1366750..e3df242 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/ResponseHeader.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/ResponseHeader.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/SubscribeRequest.java b/clients/src/main/java/org/oracle/okafka/common/requests/SubscribeRequest.java index cf9bd30..2d57ce3 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/SubscribeRequest.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/SubscribeRequest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/SubscribeResponse.java b/clients/src/main/java/org/oracle/okafka/common/requests/SubscribeResponse.java index 12afffc..5ee2acd 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/SubscribeResponse.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/SubscribeResponse.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/UnsubscribeRequest.java b/clients/src/main/java/org/oracle/okafka/common/requests/UnsubscribeRequest.java index cc53fce..83b5e80 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/UnsubscribeRequest.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/UnsubscribeRequest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/requests/UnsubscribeResponse.java b/clients/src/main/java/org/oracle/okafka/common/requests/UnsubscribeResponse.java index 97af828..2c47e47 100644 --- a/clients/src/main/java/org/oracle/okafka/common/requests/UnsubscribeResponse.java +++ b/clients/src/main/java/org/oracle/okafka/common/requests/UnsubscribeResponse.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/utils/ConnectionUtils.java b/clients/src/main/java/org/oracle/okafka/common/utils/ConnectionUtils.java index 451240e..ee223e3 100644 --- a/clients/src/main/java/org/oracle/okafka/common/utils/ConnectionUtils.java +++ b/clients/src/main/java/org/oracle/okafka/common/utils/ConnectionUtils.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** - ** Copyright (c) 2019, 2020 Oracle and/or its affiliates. + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/utils/CreateTopics.java b/clients/src/main/java/org/oracle/okafka/common/utils/CreateTopics.java index c3a6ae5..b594202 100644 --- a/clients/src/main/java/org/oracle/okafka/common/utils/CreateTopics.java +++ b/clients/src/main/java/org/oracle/okafka/common/utils/CreateTopics.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/utils/MessageIdConverter.java b/clients/src/main/java/org/oracle/okafka/common/utils/MessageIdConverter.java index ab2d82c..35a11f3 100644 --- a/clients/src/main/java/org/oracle/okafka/common/utils/MessageIdConverter.java +++ b/clients/src/main/java/org/oracle/okafka/common/utils/MessageIdConverter.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** - ** Copyright (c) 2019, 2020 Oracle and/or its affiliates. + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/main/java/org/oracle/okafka/common/utils/TNSParser.java b/clients/src/main/java/org/oracle/okafka/common/utils/TNSParser.java index 774fa16..285b10a 100644 --- a/clients/src/main/java/org/oracle/okafka/common/utils/TNSParser.java +++ b/clients/src/main/java/org/oracle/okafka/common/utils/TNSParser.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/test/java/org/oracle/okafka/clients/MockClient.java b/clients/src/test/java/org/oracle/okafka/clients/MockClient.java index 638a0cc..edfac33 100644 --- a/clients/src/test/java/org/oracle/okafka/clients/MockClient.java +++ b/clients/src/test/java/org/oracle/okafka/clients/MockClient.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ @@ -29,8 +29,10 @@ package org.oracle.okafka.clients; -//import org.apache.kafka.clients.producer.internals.ProduceResponse; -//import org.apache.kafka.clients.producer.internals.ProduceRequest; +import org.apache.kafka.clients.ClientRequest; +import org.apache.kafka.clients.ClientResponse; +import org.apache.kafka.clients.RequestCompletionHandler; +import org.apache.kafka.clients.producer.internals.ProducerMetadata; import org.apache.kafka.common.Cluster; import org.oracle.okafka.common.Node; import org.apache.kafka.common.errors.AuthenticationException; @@ -38,6 +40,8 @@ import org.oracle.okafka.common.protocol.ApiKeys; import org.oracle.okafka.common.requests.AbstractRequest; import org.oracle.okafka.common.requests.AbstractResponse; +import org.oracle.okafka.common.requests.ProduceRequest; +import org.oracle.okafka.common.requests.ProduceResponse; import org.apache.kafka.common.utils.Time; import org.oracle.okafka.test.TestCondition; import org.oracle.okafka.test.TestUtils; @@ -57,7 +61,7 @@ public class MockClient implements KafkaClient { private int correlation; private final Time time; - private final Metadata metadata; + private final ProducerMetadata metadata; private Set unavailableTopics; private Cluster cluster; private Node node = null; @@ -80,7 +84,7 @@ public MockClient(Time time) { this(time, null); } - public MockClient(Time time, Metadata metadata) { + public MockClient(Time time, ProducerMetadata metadata) { this.time = time; this.metadata = metadata; this.unavailableTopics = Collections.emptySet(); @@ -181,7 +185,7 @@ private ClientResponse dummyResponse(ClientRequest request) { if(request.apiKey() == ApiKeys.PRODUCE ) { ProduceRequest.Builder builder = (ProduceRequest.Builder)request.requestBuilder(); ProduceRequest produceRequest = builder.build(); - return new ClientResponse(request.makeHeader(), request.callback(), request.destination(), + return new ClientResponse(request.makeHeader((short)1), request.callback(), request.destination(), request.createdTimeMs(), time.milliseconds(), true, new ProduceResponse(produceRequest.getTopicpartition(), new ProduceResponse.PartitionResponse(new InvalidTopicException("This exception can be retried")))); } @@ -269,7 +273,7 @@ public ClientRequest newClientRequest(Node node, boolean expectResponse, int requestTimeoutMs, RequestCompletionHandler callback) { - return new ClientRequest(node, requestBuilder, correlation++, "mockClientId", createdTimeMs, + return new ClientRequest(""+node.id(), requestBuilder, correlation++, "mockClientId", createdTimeMs, expectResponse, requestTimeoutMs, callback); } diff --git a/clients/src/test/java/org/oracle/okafka/clients/consumer/ConsumerConfigTest.java b/clients/src/test/java/org/oracle/okafka/clients/consumer/ConsumerConfigTest.java index 782d7e9..b61964d 100644 --- a/clients/src/test/java/org/oracle/okafka/clients/consumer/ConsumerConfigTest.java +++ b/clients/src/test/java/org/oracle/okafka/clients/consumer/ConsumerConfigTest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/test/java/org/oracle/okafka/clients/consumer/ConsumerRecordTest.java b/clients/src/test/java/org/oracle/okafka/clients/consumer/ConsumerRecordTest.java index 15503f9..2c374ac 100644 --- a/clients/src/test/java/org/oracle/okafka/clients/consumer/ConsumerRecordTest.java +++ b/clients/src/test/java/org/oracle/okafka/clients/consumer/ConsumerRecordTest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/test/java/org/oracle/okafka/clients/consumer/ConsumerRecordsTest.java b/clients/src/test/java/org/oracle/okafka/clients/consumer/ConsumerRecordsTest.java index 68ea2ae..291e9d3 100644 --- a/clients/src/test/java/org/oracle/okafka/clients/consumer/ConsumerRecordsTest.java +++ b/clients/src/test/java/org/oracle/okafka/clients/consumer/ConsumerRecordsTest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/test/java/org/oracle/okafka/clients/consumer/SerializeCompatibilityOffsetAndMetadataTest.java b/clients/src/test/java/org/oracle/okafka/clients/consumer/SerializeCompatibilityOffsetAndMetadataTest.java index 40f6b00..401d9d9 100644 --- a/clients/src/test/java/org/oracle/okafka/clients/consumer/SerializeCompatibilityOffsetAndMetadataTest.java +++ b/clients/src/test/java/org/oracle/okafka/clients/consumer/SerializeCompatibilityOffsetAndMetadataTest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/test/java/org/oracle/okafka/clients/producer/KafkaProducerTest.java b/clients/src/test/java/org/oracle/okafka/clients/producer/KafkaProducerTest.java index 0b4be3e..7de47e3 100644 --- a/clients/src/test/java/org/oracle/okafka/clients/producer/KafkaProducerTest.java +++ b/clients/src/test/java/org/oracle/okafka/clients/producer/KafkaProducerTest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ @@ -30,9 +30,13 @@ package org.oracle.okafka.clients.producer; import org.oracle.okafka.clients.CommonClientConfigs; +import org.oracle.okafka.clients.KafkaClient; import org.oracle.okafka.clients.Metadata; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; //import org.oracle.okafka.clients.MockClient; import org.apache.kafka.clients.producer.internals.ProducerInterceptors; +import org.apache.kafka.clients.producer.internals.ProducerMetadata; import org.apache.kafka.common.Cluster; import org.apache.kafka.common.KafkaException; import org.oracle.okafka.common.Node; @@ -46,9 +50,12 @@ import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.serialization.ByteArraySerializer; import org.apache.kafka.common.serialization.ExtendedSerializer; +import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.common.serialization.StringSerializer; import org.oracle.okafka.common.utils.MockTime; +import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.common.utils.Time; +import org.apache.kafka.common.utils.Utils; import org.oracle.okafka.test.MockMetricsReporter; import org.oracle.okafka.test.MockPartitioner; import org.oracle.okafka.test.MockProducerInterceptor; @@ -85,6 +92,24 @@ @RunWith(PowerMockRunner.class) @PowerMockIgnore("javax.management.*") public class KafkaProducerTest { + + private static final int DEFAULT_METADATA_IDLE_MS = 5 * 60 * 1000; + + private static KafkaProducer kafkaProducer(Map configs, + Serializer keySerializer, + Serializer valueSerializer, + ProducerMetadata metadata, + KafkaClient kafkaClient, + ProducerInterceptors interceptors, + Time time) { + return new KafkaProducer<>(new ProducerConfig(ProducerConfig.appendSerializerToConfig(configs, keySerializer, valueSerializer)), + keySerializer, valueSerializer, metadata, kafkaClient, interceptors, time); +} + + private static ProducerMetadata newMetadata(long refreshBackoffMs, long expirationMs) { + return new ProducerMetadata(refreshBackoffMs, expirationMs, DEFAULT_METADATA_IDLE_MS, + new LogContext(), new ClusterResourceListeners(), Time.SYSTEM); + } @Test public void testConstructorWithSerializers() { @@ -206,27 +231,27 @@ public void testPartitionerClose() throws Exception { @Test public void shouldCloseProperlyAndThrowIfInterrupted() throws Exception { - Properties props = new Properties(); - props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999"); - props.put(ProducerConfig.ORACLE_INSTANCE_NAME, "instancename"); - props.put(ProducerConfig.ORACLE_SERVICE_NAME, "servicename"); - props.put(ProducerConfig.ORACLE_NET_TNS_ADMIN, "/temp"); - props.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG, MockPartitioner.class.getName()); - props.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, "1"); + Map configs = new HashMap<>(); + configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999"); + configs.put(ProducerConfig.ORACLE_INSTANCE_NAME, "instancename"); + configs.put(ProducerConfig.ORACLE_SERVICE_NAME, "servicename"); + configs.put(ProducerConfig.ORACLE_NET_TNS_ADMIN, "/temp"); + configs.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MockPartitioner.class.getName()); + configs.put(ProducerConfig.BATCH_SIZE_CONFIG, "1"); Time time = new MockTime(); Cluster cluster = TestUtils.singletonCluster("topic", 1); - Node node = cluster.nodes().get(0); + + Node node = (Node) cluster.nodes().get(0); - Metadata metadata = new Metadata(0, Long.MAX_VALUE, true, null); + ProducerMetadata metadata = newMetadata(0, Long.MAX_VALUE); metadata.update(cluster, Collections.emptySet(), time.milliseconds()); - MockClient client = new MockClient(time, metadata); + MockClient client = new MockClient(time,metadata); client.setNode(node); - final Producer producer = new KafkaProducer<>( - new ProducerConfig(ProducerConfig.addSerializerToConfig(props, new StringSerializer(), new StringSerializer())), - new StringSerializer(), new StringSerializer(), metadata, client); + final Producer producer = kafkaProducer( + configs,new StringSerializer(), new StringSerializer(), metadata, client,null,time); ExecutorService executor = Executors.newSingleThreadExecutor(); final AtomicReference closeException = new AtomicReference<>(); diff --git a/clients/src/test/java/org/oracle/okafka/clients/producer/ProducerRecordTest.java b/clients/src/test/java/org/oracle/okafka/clients/producer/ProducerRecordTest.java index bff2bdc..b42ef76 100644 --- a/clients/src/test/java/org/oracle/okafka/clients/producer/ProducerRecordTest.java +++ b/clients/src/test/java/org/oracle/okafka/clients/producer/ProducerRecordTest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ @@ -35,6 +35,8 @@ import static org.junit.Assert.assertFalse; import static org.junit.Assert.fail; +import org.apache.kafka.clients.producer.ProducerRecord; + public class ProducerRecordTest { @Test diff --git a/clients/src/test/java/org/oracle/okafka/clients/producer/RecordMetadataTest.java b/clients/src/test/java/org/oracle/okafka/clients/producer/RecordMetadataTest.java index ffb0a72..dfafa06 100644 --- a/clients/src/test/java/org/oracle/okafka/clients/producer/RecordMetadataTest.java +++ b/clients/src/test/java/org/oracle/okafka/clients/producer/RecordMetadataTest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ @@ -29,8 +29,9 @@ package org.oracle.okafka.clients.producer; -import org.oracle.okafka.common.TopicPartition; -import org.oracle.okafka.common.record.DefaultRecord; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.record.DefaultRecord; import org.junit.Test; import static org.junit.Assert.assertEquals; diff --git a/clients/src/test/java/org/oracle/okafka/clients/producer/RecordSendTest.java b/clients/src/test/java/org/oracle/okafka/clients/producer/RecordSendTest.java index bad116e..28667e0 100644 --- a/clients/src/test/java/org/oracle/okafka/clients/producer/RecordSendTest.java +++ b/clients/src/test/java/org/oracle/okafka/clients/producer/RecordSendTest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ @@ -44,9 +44,10 @@ import org.oracle.okafka.clients.producer.internals.FutureRecordMetadata; import org.oracle.okafka.clients.producer.internals.ProduceRequestResult; -import org.oracle.okafka.common.TopicPartition; -import org.oracle.okafka.common.errors.CorruptRecordException; -import org.oracle.okafka.common.record.RecordBatch; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.CorruptRecordException; +import org.apache.kafka.common.record.RecordBatch; +import org.apache.kafka.common.utils.Time; import org.junit.Test; public class RecordSendTest { @@ -64,7 +65,7 @@ public class RecordSendTest { public void testTimeout() throws Exception { ProduceRequestResult request = new ProduceRequestResult(topicPartition); FutureRecordMetadata future = new FutureRecordMetadata(request, relOffset, - RecordBatch.NO_TIMESTAMP, 0L, 0, 0); + RecordBatch.NO_TIMESTAMP, 0L, 0, 0,Time.SYSTEM); assertFalse("Request is not completed", future.isDone()); try { future.get(5, TimeUnit.MILLISECONDS); @@ -85,7 +86,7 @@ public void testTimeout() throws Exception { @Test(expected = ExecutionException.class) public void testError() throws Exception { FutureRecordMetadata future = new FutureRecordMetadata(asyncRequest(msgIds, new CorruptRecordException(), 50L), - relOffset, RecordBatch.NO_TIMESTAMP, 0L, 0, 0); + relOffset, RecordBatch.NO_TIMESTAMP, 0L, 0, 0,Time.SYSTEM); future.get(); } @@ -95,7 +96,7 @@ public void testError() throws Exception { @Test public void testBlocking() throws Exception { FutureRecordMetadata future = new FutureRecordMetadata(asyncRequest(msgIds, null, 50L), - relOffset, RecordBatch.NO_TIMESTAMP, 0L, 0, 0); + relOffset, RecordBatch.NO_TIMESTAMP, 0L, 0, 0,Time.SYSTEM); assertEquals((baseOffset << 16) + relOffset, future.get().offset()); } diff --git a/clients/src/test/java/org/oracle/okafka/clients/producer/internals/BufferPoolTest.java b/clients/src/test/java/org/oracle/okafka/clients/producer/internals/BufferPoolTest.java index b6b13b0..dd0ac19 100644 --- a/clients/src/test/java/org/oracle/okafka/clients/producer/internals/BufferPoolTest.java +++ b/clients/src/test/java/org/oracle/okafka/clients/producer/internals/BufferPoolTest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ @@ -29,13 +29,13 @@ package org.oracle.okafka.clients.producer.internals; -import org.oracle.okafka.common.MetricName; -import org.oracle.okafka.common.errors.TimeoutException; -import org.oracle.okafka.common.metrics.Metrics; -import org.oracle.okafka.common.metrics.Sensor; +import org.apache.kafka.common.MetricName; +import org.apache.kafka.common.errors.TimeoutException; +import org.apache.kafka.common.metrics.Metrics; +import org.apache.kafka.common.metrics.Sensor; import org.oracle.okafka.common.utils.MockTime; -import org.oracle.okafka.common.utils.Time; -import org.oracle.okafka.clients.producer.internals.BufferPool; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.clients.producer.internals.BufferPool; import org.oracle.okafka.test.TestUtils; import org.junit.After; import org.junit.Test; @@ -70,6 +70,7 @@ @RunWith(PowerMockRunner.class) public class BufferPoolTest { private final MockTime time = new MockTime(); + private final Metrics metrics = new Metrics(time); private final long maxBlockTimeMs = 2000; private final String metricGroup = "TestMetrics"; @@ -80,7 +81,7 @@ public class BufferPoolTest { public void testSimple() throws Exception { long totalMemory = 64 * 1024; int size = 1024; - BufferPool pool = new BufferPool(totalMemory, size, time); + BufferPool pool = new BufferPool(totalMemory, size, metrics, time,metricGroup); ByteBuffer buffer = pool.allocate(size, maxBlockTimeMs); assertEquals("Buffer size should equal requested size.", size, buffer.limit()); assertEquals("Unallocated memory should have shrunk", totalMemory - size, pool.unallocatedMemory()); @@ -107,7 +108,7 @@ public void testSimple() throws Exception { */ @Test(expected = IllegalArgumentException.class) public void testCantAllocateMoreMemoryThanWeHave() throws Exception { - BufferPool pool = new BufferPool(1024, 512,time); + BufferPool pool = new BufferPool(1024, 512,metrics,time,metricGroup); ByteBuffer buffer = pool.allocate(1024, maxBlockTimeMs); assertEquals(1024, buffer.limit()); pool.deallocate(buffer); @@ -119,7 +120,7 @@ public void testCantAllocateMoreMemoryThanWeHave() throws Exception { */ @Test public void testDelayedAllocation() throws Exception { - BufferPool pool = new BufferPool(5 * 1024, 1024, time); + BufferPool pool = new BufferPool(5 * 1024, 1024,metrics, time,metricGroup); ByteBuffer buffer = pool.allocate(1024, maxBlockTimeMs); CountDownLatch doDealloc = asyncDeallocate(pool, buffer); CountDownLatch allocation = asyncAllocate(pool, 5 * 1024); @@ -177,7 +178,7 @@ public void run() { */ @Test public void testBlockTimeout() throws Exception { - BufferPool pool = new BufferPool(10, 1, Time.SYSTEM); + BufferPool pool = new BufferPool(10, 1, metrics,Time.SYSTEM,metricGroup); ByteBuffer buffer1 = pool.allocate(1, maxBlockTimeMs); ByteBuffer buffer2 = pool.allocate(1, maxBlockTimeMs); ByteBuffer buffer3 = pool.allocate(1, maxBlockTimeMs); @@ -204,7 +205,7 @@ public void testBlockTimeout() throws Exception { */ @Test public void testCleanupMemoryAvailabilityWaiterOnBlockTimeout() throws Exception { - BufferPool pool = new BufferPool(2, 1, time); + BufferPool pool = new BufferPool(2, 1, metrics,time,metricGroup); pool.allocate(1, maxBlockTimeMs); try { pool.allocate(2, maxBlockTimeMs); @@ -220,7 +221,7 @@ public void testCleanupMemoryAvailabilityWaiterOnBlockTimeout() throws Exception */ @Test public void testCleanupMemoryAvailabilityWaiterOnInterruption() throws Exception { - BufferPool pool = new BufferPool(2, 1, time); + BufferPool pool = new BufferPool(2, 1, metrics, time,metricGroup); long blockTime = 5000; pool.allocate(1, maxBlockTimeMs); Thread t1 = new Thread(new BufferPoolAllocator(pool, blockTime)); @@ -280,7 +281,7 @@ public void testStressfulSituation() throws Exception { final int iterations = 50000; final int poolableSize = 1024; final long totalMemory = numThreads / 2 * poolableSize; - final BufferPool pool = new BufferPool(totalMemory, poolableSize, time); + final BufferPool pool = new BufferPool(totalMemory, poolableSize, metrics,time,metricGroup); List threads = new ArrayList(); for (int i = 0; i < numThreads; i++) threads.add(new StressTestThread(pool, iterations)); @@ -298,7 +299,7 @@ public void testLargeAvailableMemory() throws Exception { long memory = 20_000_000_000L; int poolableSize = 2_000_000_000; final AtomicInteger freeSize = new AtomicInteger(0); - BufferPool pool = new BufferPool(memory, poolableSize, time) { + BufferPool pool = new BufferPool(memory, poolableSize, metrics,time,metricGroup) { @Override protected ByteBuffer allocateByteBuffer(int size) { // Ignore size to avoid OOM due to large buffers @@ -324,7 +325,7 @@ protected int freeSize() { @Test public void outOfMemoryOnAllocation() { - BufferPool bufferPool = new BufferPool(1024, 1024, time) { + BufferPool bufferPool = new BufferPool(1024, 1024, metrics, time,metricGroup) { @Override protected ByteBuffer allocateByteBuffer(int size) { throw new OutOfMemoryError(); diff --git a/clients/src/test/java/org/oracle/okafka/clients/producer/internals/DefaultPartitionerTest.java b/clients/src/test/java/org/oracle/okafka/clients/producer/internals/DefaultPartitionerTest.java index 27e2d49..9f7670a 100644 --- a/clients/src/test/java/org/oracle/okafka/clients/producer/internals/DefaultPartitionerTest.java +++ b/clients/src/test/java/org/oracle/okafka/clients/producer/internals/DefaultPartitionerTest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ @@ -29,10 +29,11 @@ package org.oracle.okafka.clients.producer.internals; -import org.oracle.okafka.clients.producer.Partitioner; -import org.oracle.okafka.common.Cluster; +import org.apache.kafka.clients.producer.Partitioner; +import org.apache.kafka.clients.producer.internals.DefaultPartitioner; +import org.apache.kafka.common.Cluster; import org.oracle.okafka.common.Node; -import org.oracle.okafka.common.PartitionInfo; +import org.apache.kafka.common.PartitionInfo; import org.junit.Test; import java.util.Collections; diff --git a/clients/src/test/java/org/oracle/okafka/clients/producer/internals/ProducerBatchTest.java b/clients/src/test/java/org/oracle/okafka/clients/producer/internals/ProducerBatchTest.java index 1d8282c..d2a7887 100644 --- a/clients/src/test/java/org/oracle/okafka/clients/producer/internals/ProducerBatchTest.java +++ b/clients/src/test/java/org/oracle/okafka/clients/producer/internals/ProducerBatchTest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ @@ -29,19 +29,19 @@ package org.oracle.okafka.clients.producer.internals; -import org.oracle.okafka.clients.producer.Callback; -import org.oracle.okafka.clients.producer.RecordMetadata; -import org.oracle.okafka.common.KafkaException; -import org.oracle.okafka.common.TopicPartition; -import org.oracle.okafka.common.header.Header; -import org.oracle.okafka.common.header.internals.RecordHeader; -import org.oracle.okafka.common.record.CompressionType; -import org.oracle.okafka.common.record.LegacyRecord; -import org.oracle.okafka.common.record.MemoryRecords; -import org.oracle.okafka.common.record.MemoryRecordsBuilder; -import org.oracle.okafka.common.record.Record; -import org.oracle.okafka.common.record.RecordBatch; -import org.oracle.okafka.common.record.TimestampType; +import org.apache.kafka.clients.producer.Callback; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.header.Header; +import org.apache.kafka.common.header.internals.RecordHeader; +import org.apache.kafka.common.record.CompressionType; +import org.apache.kafka.common.record.LegacyRecord; +import org.apache.kafka.common.record.MemoryRecords; +import org.apache.kafka.common.record.MemoryRecordsBuilder; +import org.apache.kafka.common.record.Record; +import org.apache.kafka.common.record.RecordBatch; +import org.apache.kafka.common.record.TimestampType; import org.junit.Test; import java.nio.ByteBuffer; @@ -50,9 +50,9 @@ import java.util.Deque; import java.util.concurrent.ExecutionException; -import static org.oracle.okafka.common.record.RecordBatch.MAGIC_VALUE_V0; -import static org.oracle.okafka.common.record.RecordBatch.MAGIC_VALUE_V1; -import static org.oracle.okafka.common.record.RecordBatch.MAGIC_VALUE_V2; +import static org.apache.kafka.common.record.RecordBatch.MAGIC_VALUE_V0; +import static org.apache.kafka.common.record.RecordBatch.MAGIC_VALUE_V1; +import static org.apache.kafka.common.record.RecordBatch.MAGIC_VALUE_V2; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; diff --git a/clients/src/test/java/org/oracle/okafka/clients/producer/internals/ProducerInterceptorsTest.java b/clients/src/test/java/org/oracle/okafka/clients/producer/internals/ProducerInterceptorsTest.java index 7e728cf..cb71f97 100644 --- a/clients/src/test/java/org/oracle/okafka/clients/producer/internals/ProducerInterceptorsTest.java +++ b/clients/src/test/java/org/oracle/okafka/clients/producer/internals/ProducerInterceptorsTest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ @@ -30,11 +30,12 @@ package org.oracle.okafka.clients.producer.internals; -import org.oracle.okafka.clients.producer.ProducerInterceptor; -import org.oracle.okafka.clients.producer.ProducerRecord; -import org.oracle.okafka.clients.producer.RecordMetadata; -import org.oracle.okafka.common.KafkaException; -import org.oracle.okafka.common.TopicPartition; +import org.apache.kafka.clients.producer.ProducerInterceptor; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.clients.producer.internals.ProducerInterceptors; +import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.TopicPartition; import org.junit.Test; import java.util.ArrayList; diff --git a/clients/src/test/java/org/oracle/okafka/clients/producer/internals/RecordAccumulatorTest.java b/clients/src/test/java/org/oracle/okafka/clients/producer/internals/RecordAccumulatorTest.java index c0ce9d8..ef55481 100644 --- a/clients/src/test/java/org/oracle/okafka/clients/producer/internals/RecordAccumulatorTest.java +++ b/clients/src/test/java/org/oracle/okafka/clients/producer/internals/RecordAccumulatorTest.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ @@ -29,27 +29,27 @@ package org.oracle.okafka.clients.producer.internals; -import org.oracle.okafka.clients.producer.Callback; -import org.oracle.okafka.clients.producer.RecordMetadata; -import org.oracle.okafka.common.Cluster; -import org.oracle.okafka.common.KafkaException; +import org.apache.kafka.clients.producer.Callback; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.Cluster; +import org.apache.kafka.common.KafkaException; import org.oracle.okafka.common.Node; -import org.oracle.okafka.common.PartitionInfo; -import org.oracle.okafka.common.TopicPartition; -import org.oracle.okafka.common.metrics.Metrics; +import org.apache.kafka.common.PartitionInfo; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.metrics.Metrics; import org.oracle.okafka.common.protocol.ApiKeys; -import org.oracle.okafka.common.record.CompressionRatioEstimator; -import org.oracle.okafka.common.record.CompressionType; -import org.oracle.okafka.common.record.DefaultRecord; -import org.oracle.okafka.common.record.DefaultRecordBatch; -import org.oracle.okafka.common.record.MemoryRecords; -import org.oracle.okafka.common.record.MemoryRecordsBuilder; -import org.oracle.okafka.common.record.MutableRecordBatch; -import org.oracle.okafka.common.record.Record; -import org.oracle.okafka.common.record.TimestampType; -import org.oracle.okafka.common.utils.LogContext; +import org.apache.kafka.common.record.CompressionRatioEstimator; +import org.apache.kafka.common.record.CompressionType; +import org.apache.kafka.common.record.DefaultRecord; +import org.apache.kafka.common.record.DefaultRecordBatch; +import org.apache.kafka.common.record.MemoryRecords; +import org.apache.kafka.common.record.MemoryRecordsBuilder; +import org.apache.kafka.common.record.MutableRecordBatch; +import org.apache.kafka.common.record.Record; +import org.apache.kafka.common.record.TimestampType; +import org.apache.kafka.common.utils.LogContext; import org.oracle.okafka.common.utils.MockTime; -import org.oracle.okafka.common.utils.Time; +import org.apache.kafka.common.utils.Time; import org.oracle.okafka.test.TestUtils; import org.junit.After; import org.junit.Test; @@ -117,7 +117,7 @@ public void testFull() throws Exception { int appends = expectedNumAppends(batchSize); for (int i = 0; i < appends; i++) { // append to the first batch - accum.append(tp1, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs); + accum.append(tp1, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs,true,time.milliseconds()); Deque partitionBatches = accum.batches().get(tp1); assertEquals(1, partitionBatches.size()); diff --git a/clients/src/test/java/org/oracle/okafka/common/utils/MockTime.java b/clients/src/test/java/org/oracle/okafka/common/utils/MockTime.java index 748a5e6..6ac5d56 100644 --- a/clients/src/test/java/org/oracle/okafka/common/utils/MockTime.java +++ b/clients/src/test/java/org/oracle/okafka/common/utils/MockTime.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ @@ -32,20 +32,24 @@ import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Supplier; + +import org.apache.kafka.common.errors.TimeoutException; +import org.apache.kafka.common.utils.Time; /** * A clock that you can manually advance by calling sleep */ public class MockTime implements Time { - interface MockTimeListener { - void tick(); + public interface Listener { + void onTimeUpdated(); } /** * Listeners which are waiting for time changes. */ - private final CopyOnWriteArrayList listeners = new CopyOnWriteArrayList<>(); + private final CopyOnWriteArrayList listeners = new CopyOnWriteArrayList<>(); private final long autoTickMs; @@ -68,7 +72,7 @@ public MockTime(long autoTickMs, long currentTimeMs, long currentHighResTimeNs) this.autoTickMs = autoTickMs; } - public void addListener(MockTimeListener listener) { + public void addListener(Listener listener) { listeners.add(listener); } @@ -84,11 +88,6 @@ public long nanoseconds() { return highResTimeNs.get(); } - @Override - public long hiResClockMs() { - return TimeUnit.NANOSECONDS.toMillis(nanoseconds()); - } - private void maybeSleep(long ms) { if (ms != 0) sleep(ms); @@ -101,6 +100,27 @@ public void sleep(long ms) { tick(); } + @Override + public void waitObject(Object obj, Supplier condition, long deadlineMs) throws InterruptedException { + Listener listener = () -> { + synchronized (obj) { + obj.notify(); + } + }; + listeners.add(listener); + try { + synchronized (obj) { + while (milliseconds() < deadlineMs && !condition.get()) { + obj.wait(); + } + if (!condition.get()) + throw new TimeoutException("Condition not satisfied before deadline"); + } + } finally { + listeners.remove(listener); + } + } + public void setCurrentTimeMs(long newMs) { long oldMs = timeMs.getAndSet(newMs); @@ -113,8 +133,8 @@ public void setCurrentTimeMs(long newMs) { } private void tick() { - for (MockTimeListener listener : listeners) { - listener.tick(); + for (Listener listener : listeners) { + listener.onTimeUpdated(); } } -} +} \ No newline at end of file diff --git a/clients/src/test/java/org/oracle/okafka/common/utils/Serializer.java b/clients/src/test/java/org/oracle/okafka/common/utils/Serializer.java index fefa2c5..e3d62a0 100644 --- a/clients/src/test/java/org/oracle/okafka/common/utils/Serializer.java +++ b/clients/src/test/java/org/oracle/okafka/common/utils/Serializer.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/test/java/org/oracle/okafka/test/MockMetricsReporter.java b/clients/src/test/java/org/oracle/okafka/test/MockMetricsReporter.java index 67fee39..4659b62 100644 --- a/clients/src/test/java/org/oracle/okafka/test/MockMetricsReporter.java +++ b/clients/src/test/java/org/oracle/okafka/test/MockMetricsReporter.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ @@ -29,8 +29,8 @@ package org.oracle.okafka.test; -import org.oracle.okafka.common.metrics.KafkaMetric; -import org.oracle.okafka.common.metrics.MetricsReporter; +import org.apache.kafka.common.metrics.KafkaMetric; +import org.apache.kafka.common.metrics.MetricsReporter; import java.util.List; import java.util.Map; diff --git a/clients/src/test/java/org/oracle/okafka/test/MockPartitioner.java b/clients/src/test/java/org/oracle/okafka/test/MockPartitioner.java index df3cc25..b960040 100644 --- a/clients/src/test/java/org/oracle/okafka/test/MockPartitioner.java +++ b/clients/src/test/java/org/oracle/okafka/test/MockPartitioner.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ @@ -29,8 +29,8 @@ package org.oracle.okafka.test; -import org.oracle.okafka.common.Cluster; -import org.oracle.okafka.clients.producer.Partitioner; +import org.apache.kafka.common.Cluster; +import org.apache.kafka.clients.producer.Partitioner; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; diff --git a/clients/src/test/java/org/oracle/okafka/test/MockProducerInterceptor.java b/clients/src/test/java/org/oracle/okafka/test/MockProducerInterceptor.java index b4e917c..6792e53 100644 --- a/clients/src/test/java/org/oracle/okafka/test/MockProducerInterceptor.java +++ b/clients/src/test/java/org/oracle/okafka/test/MockProducerInterceptor.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ @@ -31,12 +31,12 @@ package org.oracle.okafka.test; import org.oracle.okafka.clients.producer.ProducerConfig; -import org.oracle.okafka.clients.producer.ProducerInterceptor; -import org.oracle.okafka.clients.producer.ProducerRecord; -import org.oracle.okafka.clients.producer.RecordMetadata; -import org.oracle.okafka.common.ClusterResourceListener; -import org.oracle.okafka.common.ClusterResource; -import org.oracle.okafka.common.config.ConfigException; +import org.apache.kafka.clients.producer.ProducerInterceptor; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.ClusterResourceListener; +import org.apache.kafka.common.ClusterResource; +import org.apache.kafka.common.config.ConfigException; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; diff --git a/clients/src/test/java/org/oracle/okafka/test/MockSerializer.java b/clients/src/test/java/org/oracle/okafka/test/MockSerializer.java index f2bd263..a9ec04e 100644 --- a/clients/src/test/java/org/oracle/okafka/test/MockSerializer.java +++ b/clients/src/test/java/org/oracle/okafka/test/MockSerializer.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ @@ -29,9 +29,9 @@ package org.oracle.okafka.test; -import org.oracle.okafka.common.ClusterResourceListener; -import org.oracle.okafka.common.ClusterResource; -import org.oracle.okafka.common.serialization.Serializer; +import org.apache.kafka.common.ClusterResourceListener; +import org.apache.kafka.common.ClusterResource; +import org.apache.kafka.common.serialization.Serializer; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; diff --git a/clients/src/test/java/org/oracle/okafka/test/TestCondition.java b/clients/src/test/java/org/oracle/okafka/test/TestCondition.java index d4ac79a..f9d8a0b 100644 --- a/clients/src/test/java/org/oracle/okafka/test/TestCondition.java +++ b/clients/src/test/java/org/oracle/okafka/test/TestCondition.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/clients/src/test/java/org/oracle/okafka/test/TestUtils.java b/clients/src/test/java/org/oracle/okafka/test/TestUtils.java index c2fe592..9f38ac7 100644 --- a/clients/src/test/java/org/oracle/okafka/test/TestUtils.java +++ b/clients/src/test/java/org/oracle/okafka/test/TestUtils.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ @@ -31,10 +31,10 @@ import org.oracle.okafka.clients.consumer.ConsumerConfig; import org.oracle.okafka.clients.producer.ProducerConfig; -import org.oracle.okafka.common.Cluster; +import org.apache.kafka.common.Cluster; import org.oracle.okafka.common.Node; -import org.oracle.okafka.common.PartitionInfo; -import org.oracle.okafka.common.utils.Utils; +import org.apache.kafka.common.PartitionInfo; +import org.apache.kafka.common.utils.Utils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/clients/src/test/java/org/oracle/okafka/tests/ConsumerMetricsTest.java b/clients/src/test/java/org/oracle/okafka/tests/ConsumerMetricsTest.java new file mode 100644 index 0000000..0c3e74b --- /dev/null +++ b/clients/src/test/java/org/oracle/okafka/tests/ConsumerMetricsTest.java @@ -0,0 +1,137 @@ +package org.oracle.okafka.tests; + +import java.io.File; +import java.io.FileWriter; +import java.io.IOException; +import java.time.Duration; +import java.util.Arrays; +import java.util.Collection; +import java.util.Map; +import java.util.Properties; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.common.Metric; +import org.apache.kafka.common.MetricName; +import org.apache.kafka.common.TopicPartition; +import org.junit.Test; +import org.oracle.okafka.clients.consumer.KafkaConsumer; + +public class ConsumerMetricsTest { + public static void getMetricData(Consumer consumer,String fileName) { + try { + Map metricData = consumer.metrics(); + File csvFile = new File(System.getProperty("user.dir") +fileName+ ".csv"); + FileWriter fileWriter = new FileWriter(csvFile); + StringBuilder headLine = new StringBuilder(); + headLine.append("Name"); + headLine.append(','); + headLine.append("Group"); + headLine.append(','); + headLine.append("Description"); + headLine.append(','); + headLine.append("Tags"); + headLine.append(','); + headLine.append("Value"); + headLine.append("\n"); + fileWriter.write(headLine.toString()); + metricData.forEach((a, b) -> { + try { + + StringBuilder line = new StringBuilder(); + + line.append(a.name()); + line.append(','); + line.append(a.group()); + line.append(','); + line.append(a.description()); + if(a.tags().containsKey("node-id") || a.tags().containsKey("topic")) { + if(a.tags().containsKey("node-id")) { + line.append(','); + line.append(a.tags().get("node-id")); + } + if(a.tags().containsKey("topic")) { + line.append(','); + line.append("topic-"+a.tags().get("topic")); + } + }else{ + line.append(','); + line.append(""); + + } + line.append(','); + line.append(b.metricValue().toString()); + line.append("\n"); + fileWriter.write(line.toString()); + + } catch (IOException e) { + e.printStackTrace(); + } + + }); + fileWriter.close(); + } catch(IOException e) { + e.printStackTrace(); + } + } + + + + @Test + public void ConsumingTest() { + Properties prop = new Properties(); + prop = OkafkaSetup.setup(); + prop.put("group.id" , "S1"); + prop.put("max.poll.records", 1000); + prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + prop.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + Consumer consumer = new KafkaConsumer(prop); + consumer.subscribe(Arrays.asList("TEQ")); + int expectedMsgCnt =1000; + int retryCount=1; + int msgCnt = 0; + try { + while(true) { + try { + ConsumerRecords records = consumer.poll(Duration.ofMillis(10000)); + + Collection partitions = records.partitions(); + for (ConsumerRecord record : records) + System.out.printf("partition = %d, offset = %d, key = %s, value =%s\n ", record.partition(), record.offset(), record.key(), record.value()); + + if(records != null && records.count() > 0) { + msgCnt += records.count(); + System.out.println("Committing records " + records.count()); + consumer.commitSync(); + + if(msgCnt >= expectedMsgCnt ) + { + System.out.println("Received " + msgCnt + " Expected " + expectedMsgCnt +". Exiting Now."); + break; + } + } + else { + if(retryCount>3) + break; + System.out.println("No Record Fetched. Retrying in 1 second"); + Thread.sleep(1000); + retryCount++; + } + }catch(Exception e) + { + throw e; + } + } + + }catch(Exception e) + { + System.out.println("Exception from consumer " + e); + e.printStackTrace(); + }finally { + ConsumerMetricsTest.getMetricData(consumer, "afterConsumingOkafka"); + System.out.println("Closing Consumer"); + consumer.close(); + } + } +} diff --git a/clients/src/test/java/org/oracle/okafka/tests/OkafkaAutoOffsetReset.java b/clients/src/test/java/org/oracle/okafka/tests/OkafkaAutoOffsetReset.java new file mode 100644 index 0000000..47786b8 --- /dev/null +++ b/clients/src/test/java/org/oracle/okafka/tests/OkafkaAutoOffsetReset.java @@ -0,0 +1,76 @@ +package org.oracle.okafka.tests; + + +import java.io.IOException; +import java.time.Duration; +import java.util.Arrays; +import java.util.Collection; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.common.TopicPartition; +import org.junit.Test; +import org.oracle.okafka.clients.consumer.KafkaConsumer; +import org.oracle.okafka.clients.consumer.internals.SubscriptionState; + +public class OkafkaAutoOffsetReset { + + @Test + public void autoOffsetSeekTest() throws IOException { + Properties prop = new Properties(); + prop = OkafkaSetup.setup(); + prop.put("group.id" , "S1"); + prop.put("max.poll.records", 1000); + prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + prop.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + prop.put("auto.offset.reset", "earliest"); + + Consumer consumer = new KafkaConsumer(prop); + + consumer.subscribe(Arrays.asList("TEQ")); + + int expectedMsgCnt = 1000; + int msgCnt = 0; + try { + while(true) { + try { + ConsumerRecords records = consumer.poll(Duration.ofMillis(10000)); + + for (ConsumerRecord record : records) + System.out.printf("partition = %d, offset = %d, key = %s, value =%s\n ", record.partition(), record.offset(), record.key(), record.value()); + + if(records != null && records.count() > 0) { + msgCnt += records.count(); + System.out.println("Committing records " + records.count()); + consumer.commitSync(); + + if(msgCnt >= expectedMsgCnt ) + { + System.out.println("Received " + msgCnt + " Expected " + expectedMsgCnt +". Exiting Now."); + break; + } + } + else { + System.out.println("No Record Fetched. Retrying in 1 second"); + Thread.sleep(1000); + } + }catch(Exception e) + { + throw e; + } + } + }catch(Exception e) + { + System.out.println("Exception from consumer " + e); + e.printStackTrace(); + }finally { + System.out.println("Closing Consumer"); + consumer.close(); + } + } + } diff --git a/clients/src/test/java/org/oracle/okafka/tests/OkafkaDeleteTopic.java b/clients/src/test/java/org/oracle/okafka/tests/OkafkaDeleteTopic.java new file mode 100644 index 0000000..4828ed7 --- /dev/null +++ b/clients/src/test/java/org/oracle/okafka/tests/OkafkaDeleteTopic.java @@ -0,0 +1,39 @@ +package org.oracle.okafka.tests; +import java.util.Arrays; +import java.util.Collections; +import java.util.Properties; +import java.util.concurrent.ExecutionException; +import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.admin.CreateTopicsResult; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.common.KafkaFuture; +import org.junit.Test; +import org.oracle.okafka.clients.admin.AdminClient; +import org.oracle.okafka.clients.admin.DeleteTopicsResult; +import org.oracle.okafka.clients.admin.KafkaAdminClient; + +public class OkafkaDeleteTopic { + + @Test + public void DeleteTopicTest() { + try (Admin admin = AdminClient.create(OkafkaSetup.setup())) { + KafkaAdminClient kAdminClient = (((org.oracle.okafka.clients.admin.KafkaAdminClient)admin)); + DeleteTopicsResult delResult = kAdminClient.deleteTopics(Collections.singletonList("TEQ"), new org.oracle.okafka.clients.admin.DeleteTopicsOptions()); + try { + KafkaFuture ftr = delResult.all(); + ftr.get(); + System.out.println("Main Thread Out of wait now"); + } catch ( InterruptedException | ExecutionException e ) { + + throw new IllegalStateException(e); + } + System.out.println("Auto Closing admin now"); + } + catch(Exception e) + { + System.out.println("Exception while creating topic " + e); + e.printStackTrace(); + } + System.out.println("Main thread completed "); + } +} \ No newline at end of file diff --git a/clients/src/test/java/org/oracle/okafka/tests/OkafkaSeekToBeginning.java b/clients/src/test/java/org/oracle/okafka/tests/OkafkaSeekToBeginning.java new file mode 100644 index 0000000..2c8aeec --- /dev/null +++ b/clients/src/test/java/org/oracle/okafka/tests/OkafkaSeekToBeginning.java @@ -0,0 +1,97 @@ +package org.oracle.okafka.tests; + +import java.io.IOException; +import java.time.Duration; +import java.util.Arrays; +import java.util.Collection; +import java.util.Properties; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.common.TopicPartition; +import org.junit.Test; +import org.oracle.okafka.clients.consumer.KafkaConsumer; + +public class OkafkaSeekToBeginning { + @Test + public void SeekBeginningTest() throws IOException { + Properties prop = new Properties(); + prop = OkafkaSetup.setup(); + prop.put("group.id" , "S1"); + prop.put("max.poll.records", 1000); + prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + prop.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + + Consumer consumer = new KafkaConsumer(prop); + try { + + consumer.subscribe(Arrays.asList("TEQ"), new ConsumerRebalanceListener() { + @Override + public synchronized void onPartitionsRevoked(Collection partitions) { + System.out.println("Partitions revoked for rebalance."); + } + @Override + public synchronized void onPartitionsAssigned(Collection partitions) { + System.out.println("New Partitions assigned after rebalance"); + try { + consumer.seekToBeginning(partitions); + } + catch (Exception e) { + e.printStackTrace(); + } + } + }); + } + catch(Exception e) { + System.out.println(e); + e.printStackTrace(); + } + int expectedMsgCnt = 1000; + int msgCnt = 0; + try { + + while(true) { + try { + ConsumerRecords records = consumer.poll(Duration.ofMillis(10000)); + + for (ConsumerRecord record : records) + System.out.printf("partition = %d, offset = %d, key = %s, value =%s\n ", record.partition(), record.offset(), record.key(), record.value()); + + if(records != null && records.count() > 0) { + msgCnt += records.count(); + System.out.println("Committing records " + records.count()); + consumer.commitSync(); + + if(msgCnt >= expectedMsgCnt ) + { + System.out.println("Received " + msgCnt + " Expected " + expectedMsgCnt +". Exiting Now."); + break; + } + } + else { + System.out.println("No Record Fetched. Retrying in 1 second"); + Thread.sleep(1000); + } + + }catch(Exception e) + { + throw e; + } + } + + }catch(Exception e) + { + System.out.println("Exception from consumer " + e); + e.printStackTrace(); + }finally { + System.out.println("Closing Consumer"); + consumer.close(); + } + } + + } + + + + diff --git a/clients/src/test/java/org/oracle/okafka/tests/OkafkaSeekToEnd.java b/clients/src/test/java/org/oracle/okafka/tests/OkafkaSeekToEnd.java new file mode 100644 index 0000000..6f8a9a8 --- /dev/null +++ b/clients/src/test/java/org/oracle/okafka/tests/OkafkaSeekToEnd.java @@ -0,0 +1,99 @@ +package org.oracle.okafka.tests; + +import java.io.IOException; +import java.time.Duration; +import java.time.Instant; +import java.util.Arrays; +import java.util.Collection; +import java.util.Properties; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.common.TopicPartition; +import org.junit.Test; +import org.oracle.okafka.clients.consumer.KafkaConsumer; + +public class OkafkaSeekToEnd { + @Test + public void SeekEndTest() throws IOException { + Properties prop = new Properties(); + prop = OkafkaSetup.setup(); + prop.put("group.id" , "S1"); + prop.put("max.poll.records", 1000); + prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + prop.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + + Consumer consumer = new KafkaConsumer(prop); + try { + consumer.subscribe(Arrays.asList("TEQ"), new ConsumerRebalanceListener() { + @Override + public synchronized void onPartitionsRevoked(Collection partitions) { + System.out.println("Partitions revoked for rebalance."); + } + @Override + public synchronized void onPartitionsAssigned(Collection partitions) { + System.out.println("New Partitions assigned after rebalance"); + try { + consumer.seekToEnd(partitions); + } + catch (Exception e) { + e.printStackTrace(); + } + } + }); + } + catch(Exception e) { + System.out.println(e); + e.printStackTrace(); + } + int expectedMsgCnt = 1000; + int msgCnt = 0; + try { + Instant starttime = Instant.now(); + long runtime =0; + while(true && runtime <=120) { + try { + ConsumerRecords records = consumer.poll(Duration.ofMillis(10000)); + + for (ConsumerRecord record : records) + System.out.printf("partition = %d, offset = %d, key = %s, value =%s\n ", record.partition(), record.offset(), record.key(), record.value()); + + if(records != null && records.count() > 0) { + msgCnt += records.count(); + System.out.println("Committing records " + records.count()); + consumer.commitSync(); + + if(msgCnt >= expectedMsgCnt ) + { + System.out.println("Received " + msgCnt + " Expected " + expectedMsgCnt +". Exiting Now."); + break; + } + } + else { + System.out.println("No Record Fetched. Retrying in 1 second"); + Thread.sleep(1000); + } + runtime = Duration.between(starttime, Instant.now()).toSeconds(); + + }catch(Exception e) + { + throw e; + } + } + + }catch(Exception e) + { + System.out.println("Exception from consumer " + e); + e.printStackTrace(); + }finally { + System.out.println("Closing Consumer"); + consumer.close(); + } + } + + } + + + + diff --git a/clients/src/test/java/org/oracle/okafka/tests/OkafkaSetup.java b/clients/src/test/java/org/oracle/okafka/tests/OkafkaSetup.java new file mode 100644 index 0000000..cdedf19 --- /dev/null +++ b/clients/src/test/java/org/oracle/okafka/tests/OkafkaSetup.java @@ -0,0 +1,26 @@ +package org.oracle.okafka.tests; + +import java.io.FileInputStream; +import java.io.InputStream; +import java.util.Properties; +import org.junit.BeforeClass; + +public class OkafkaSetup { + + @BeforeClass + public static Properties setup(){ + + final Properties BaseProperties = new Properties(); + InputStream input; + try { + input = new FileInputStream("config.properties"); + BaseProperties.load(input); + } catch (Exception e) { + System.out.println("Exception whlie loading config.properties file. " + e); + e.printStackTrace(); + } + return BaseProperties; + } + + +} diff --git a/clients/src/test/java/org/oracle/okafka/tests/OkafkaUnsubscribe.java b/clients/src/test/java/org/oracle/okafka/tests/OkafkaUnsubscribe.java new file mode 100644 index 0000000..88cddd3 --- /dev/null +++ b/clients/src/test/java/org/oracle/okafka/tests/OkafkaUnsubscribe.java @@ -0,0 +1,106 @@ +package org.oracle.okafka.tests; + +import org.junit.Test; +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; +import java.time.Duration; +import java.util.*; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.common.TopicPartition; +import org.oracle.okafka.clients.consumer.KafkaConsumer; +import org.oracle.okafka.clients.consumer.internals.SubscriptionState; + +public class OkafkaUnsubscribe{ + + @Test + public void UnsubscribeTest() throws IOException { + Properties prop = new Properties(); + prop = OkafkaSetup.setup(); + prop.put("group.id" , "S1"); + prop.put("max.poll.records", 1000); + prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + prop.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + Consumer consumer = new KafkaConsumer(prop); + consumer.subscribe(Arrays.asList("TEQ")); + int expectedMsgCnt = 1000; + int msgCnt = 0; + try { + while(true) { + try { + ConsumerRecords records = consumer.poll(Duration.ofMillis(10000)); + Collection partitions = records.partitions(); + for (ConsumerRecord record : records) + System.out.printf("partition = %d, offset = %d, key = %s, value =%s\n ", record.partition(), record.offset(), record.key(), record.value()); + + if(records != null && records.count() > 0) { + msgCnt += records.count(); + System.out.println("Committing records " + records.count()); + consumer.commitSync(); + + if(msgCnt >= expectedMsgCnt ) + { + System.out.println("Received " + msgCnt + " Expected " + expectedMsgCnt +". Exiting Now."); + break; + } + } + else { + System.out.println("No Record Fetched. Retrying in 1 second"); + Thread.sleep(1000); + } + }catch(Exception e) + { + throw e; + } + } + try { + consumer.unsubscribe(); + } + catch(Exception e) { + System.out.println("Exception while unsubscribe" + e); + e.printStackTrace(); + } + + while(true) { + try { + ConsumerRecords records = consumer.poll(Duration.ofMillis(10000)); + Collection partitions = records.partitions(); + for (ConsumerRecord record : records) + System.out.printf("partition = %d, offset = %d, key = %s, value =%s\n ", record.partition(), record.offset(), record.key(), record.value()); + + if(records != null && records.count() > 0) { + msgCnt += records.count(); + System.out.println("Committing records " + records.count()); + consumer.commitSync(); + + if(msgCnt >= expectedMsgCnt ) + { + System.out.println("Received " + msgCnt + " Expected " + expectedMsgCnt +". Exiting Now."); + break; + } + } + else { + System.out.println("No Record Fetched. Retrying in 1 second"); + Thread.sleep(1000); + } + }catch(Exception e) + { + throw e; + } + } + + + }catch(Exception e) + { + System.out.println("Exception from consumer " + e); + e.printStackTrace(); + }finally { + System.out.println("Closing Consumer"); + consumer.close(); + } + } + } \ No newline at end of file diff --git a/clients/src/test/java/org/oracle/okafka/tests/ProducerMetricsTest.java b/clients/src/test/java/org/oracle/okafka/tests/ProducerMetricsTest.java new file mode 100644 index 0000000..12abf6b --- /dev/null +++ b/clients/src/test/java/org/oracle/okafka/tests/ProducerMetricsTest.java @@ -0,0 +1,112 @@ +package org.oracle.okafka.tests; + +import org.junit.Test; +import org.oracle.okafka.clients.producer.KafkaProducer; + + +import java.lang.Thread; +import java.io.File; +import java.io.FileWriter; +import java.io.IOException; +import java.util.Map; +import java.util.Properties; +import java.util.concurrent.Future; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.Metric; +import org.apache.kafka.common.MetricName; + +public class ProducerMetricsTest{ + + public static void getMetricData(Producer producer,String fileName) { + try { + Map metricData = producer.metrics(); + File csvFile = new File(System.getProperty("user.dir")+fileName+ ".csv"); + FileWriter fileWriter = new FileWriter(csvFile); + StringBuilder headLine = new StringBuilder(); + headLine.append("Name"); + headLine.append(','); + headLine.append("Group"); + headLine.append(','); + headLine.append("Description"); + headLine.append(','); + headLine.append("Tags"); + headLine.append(','); + headLine.append("Value"); + headLine.append("\n"); + fileWriter.write(headLine.toString()); + metricData.forEach((a, b) -> { + try { + + StringBuilder line = new StringBuilder(); + + line.append(a.name()); + line.append(','); + line.append(a.group()); + line.append(','); + line.append(a.description()); + if(a.tags().containsKey("node-id") || a.tags().containsKey("topic")) { + if(a.tags().containsKey("node-id")) { + line.append(','); + line.append(a.tags().get("node-id")); + } + if(a.tags().containsKey("topic")) { + line.append(','); + line.append("topic-"+a.tags().get("topic")); + } + }else{ + line.append(','); + line.append(""); + + } + + + + line.append(','); + line.append(b.metricValue().toString()); + + line.append("\n"); + fileWriter.write(line.toString()); + + } catch (IOException e) { + e.printStackTrace(); + } + + }); + fileWriter.close(); + } catch(IOException e) { + e.printStackTrace(); + } + } + + @Test + public void ProducerTest() { + try { + + Properties prop = new Properties(); + prop = OkafkaSetup.setup(); + prop.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + prop.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + Producer producer = new KafkaProducer(prop); + + int msgCnt = 100; + for(int i=0;i producerRecord = new ProducerRecord("TEQ", i+"", "Test message # " + i); + producer.send(producerRecord); + } + System.out.println("Produced "+ msgCnt +" messages."); + + Thread.sleep(9000); + ProducerMetricsTest.getMetricData(producer,"afterProducingOkafka"); + + producer.close(); + System.out.println("producer closed"); + } + catch(Exception e) + { + System.out.println("Exception in Main " + e ); + e.printStackTrace(); + } + } +} diff --git a/clients/src/test/java/org/oracle/okafka/tests/SimpleOkafkaAdmin.java b/clients/src/test/java/org/oracle/okafka/tests/SimpleOkafkaAdmin.java new file mode 100644 index 0000000..1d5b588 --- /dev/null +++ b/clients/src/test/java/org/oracle/okafka/tests/SimpleOkafkaAdmin.java @@ -0,0 +1,41 @@ +package org.oracle.okafka.tests; + +import java.util.Arrays; +import java.util.Properties; +import java.util.concurrent.ExecutionException; + +import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.admin.CreateTopicsResult; +import org.apache.kafka.clients.admin.CreateTopicsResult.TopicMetadataAndConfig; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.common.KafkaFuture; +import org.junit.Test; +import org.oracle.okafka.clients.admin.AdminClient; + +public class SimpleOkafkaAdmin { + + @Test + public void AdminTest() { + try (Admin admin = AdminClient.create(OkafkaSetup.setup())) { + CreateTopicsResult result = admin.createTopics(Arrays.asList( + new NewTopic("TEQ",5, (short)1))); + try { + KafkaFuture ftr = result.all(); + ftr.get(); + System.out.println("Main Thread Out of wait now"); + } catch ( InterruptedException | ExecutionException e ) { + + throw new IllegalStateException(e); + } + System.out.println("Auto Closing admin now"); + } + catch(Exception e) + { + System.out.println("Exception while creating topic " + e); + e.printStackTrace(); + } + + System.out.println("Main thread complete "); + + } +} \ No newline at end of file diff --git a/clients/src/test/java/org/oracle/okafka/tests/SimpleOkafkaConsumer.java b/clients/src/test/java/org/oracle/okafka/tests/SimpleOkafkaConsumer.java new file mode 100644 index 0000000..f3c50e5 --- /dev/null +++ b/clients/src/test/java/org/oracle/okafka/tests/SimpleOkafkaConsumer.java @@ -0,0 +1,69 @@ +package org.oracle.okafka.tests; + +import org.junit.Test; +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; +import java.time.Duration; +import java.util.*; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.common.TopicPartition; +import org.oracle.okafka.clients.consumer.KafkaConsumer; +import org.oracle.okafka.clients.consumer.internals.SubscriptionState; + +public class SimpleOkafkaConsumer{ + + @Test + public void ConsumerTest() { + Properties prop = new Properties(); + prop = OkafkaSetup.setup(); + prop.put("group.id" , "S1"); + prop.put("max.poll.records", 1000); + prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + prop.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + Consumer consumer = new KafkaConsumer(prop); + consumer.subscribe(Arrays.asList("TEQ")); + int expectedMsgCnt = 1000; + int msgCnt = 0; + try { + while(true) { + try { + ConsumerRecords records = consumer.poll(Duration.ofMillis(10000)); + Collection partitions = records.partitions(); + for (ConsumerRecord record : records) + System.out.printf("partition = %d, offset = %d, key = %s, value =%s\n ", record.partition(), record.offset(), record.key(), record.value()); + + if(records != null && records.count() > 0) { + msgCnt += records.count(); + System.out.println("Committing records " + records.count()); + consumer.commitSync(); + + if(msgCnt >= expectedMsgCnt ) + { + System.out.println("Received " + msgCnt + " Expected " + expectedMsgCnt +". Exiting Now."); + break; + } + } + else { + System.out.println("No Record Fetched. Retrying in 1 second"); + Thread.sleep(1000); + } + }catch(Exception e) + { + throw e; + } + } + }catch(Exception e) + { + System.out.println("Exception from consumer " + e); + e.printStackTrace(); + }finally { + System.out.println("Closing Consumer"); + consumer.close(); + } + } + } diff --git a/clients/src/test/java/org/oracle/okafka/tests/SimpleOkafkaProducer.java b/clients/src/test/java/org/oracle/okafka/tests/SimpleOkafkaProducer.java new file mode 100644 index 0000000..c66a73e --- /dev/null +++ b/clients/src/test/java/org/oracle/okafka/tests/SimpleOkafkaProducer.java @@ -0,0 +1,42 @@ +package org.oracle.okafka.tests; + +import org.junit.BeforeClass; +import org.junit.Test; +import org.oracle.okafka.clients.producer.KafkaProducer; +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; +import java.util.Properties; +import java.util.concurrent.Future; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; + +public class SimpleOkafkaProducer{ + + @Test + public void ProducerTest() { + try { + Properties prop = OkafkaSetup.setup(); + prop.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + prop.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + Producer producer = new KafkaProducer(prop); + Future lastFuture = null; + int msgCnt = 1000; + for(int i=0;i producerRecord = new ProducerRecord("TEQ", i+"", "Test message # " + i); + lastFuture =producer.send(producerRecord); + } + System.out.println("Produced "+ msgCnt +" messages."); + lastFuture.get(); + producer.close(); + } + catch(Exception e) + { + System.out.println("Exception in Main " + e ); + e.printStackTrace(); + } + } +} diff --git a/clients/src/test/java/org/oracle/okafka/tests/TestRunner.java b/clients/src/test/java/org/oracle/okafka/tests/TestRunner.java new file mode 100644 index 0000000..71703c2 --- /dev/null +++ b/clients/src/test/java/org/oracle/okafka/tests/TestRunner.java @@ -0,0 +1,26 @@ +package org.oracle.okafka.tests; +import java.util.Scanner; + +import org.junit.runner.JUnitCore; +import org.junit.runner.Result; +import org.junit.runner.notification.Failure; + + +class TestRunner { + + public static void main(String[] args) { + + Result result = new Result(); + + result = JUnitCore.runClasses(SimpleOkafkaAdmin.class, SimpleOkafkaProducer.class, OkafkaAutoOffsetReset.class, + SimpleOkafkaProducer.class, OkafkaSeekToEnd.class, OkafkaSeekToBeginning.class, SimpleOkafkaProducer.class, + OkafkaUnsubscribe.class, OkafkaDeleteTopic.class,ProducerMetricsTest.class, ConsumerMetricsTest.class); + + for (Failure failure : result.getFailures()) { + System.out.println("Test failure : "+ failure.toString()); + } + System.out.println("Tests ran succesfully: " + result.wasSuccessful()); + + } + } + diff --git a/examples/consumer/src/main/java/org/oracle/okafka/examples/Consumer.java b/examples/consumer/src/main/java/org/oracle/okafka/examples/Consumer.java index 5e8f2af..887bb77 100644 --- a/examples/consumer/src/main/java/org/oracle/okafka/examples/Consumer.java +++ b/examples/consumer/src/main/java/org/oracle/okafka/examples/Consumer.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** - ** Copyright (c) 2019, 2020 Oracle and/or its affiliates. + ** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ package org.oracle.okafka.examples; diff --git a/examples/producer/src/main/java/org/oracle/okafka/examples/Producer.java b/examples/producer/src/main/java/org/oracle/okafka/examples/Producer.java index fb84bf5..eaaa967 100644 --- a/examples/producer/src/main/java/org/oracle/okafka/examples/Producer.java +++ b/examples/producer/src/main/java/org/oracle/okafka/examples/Producer.java @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */ diff --git a/settings.gradle b/settings.gradle index 0ffadb0..767073f 100644 --- a/settings.gradle +++ b/settings.gradle @@ -1,7 +1,7 @@ /* ** OKafka Java Client version 23.4. ** -** Copyright (c) 2019, 2020 Oracle and/or its affiliates. +** Copyright (c) 2019, 2024 Oracle and/or its affiliates. ** Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl. */